merge: 인스타 슬레이트 패키지 다운로드 버튼

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.gitignore

@@ -26,3 +26,6 @@ dist-ssr
 
 # Superpowers visual companion (mockup files)
 .superpowers/
+
+# git worktrees (subagent-driven 격리 작업)
+.worktrees/

CHECK_POINT.md

@@ -0,0 +1,109 @@
+# web-ui CHECK_POINT
+
+> React 18 + Vite + react-router-dom v6. Dev port 3007. NAS Docker 백엔드의 프론트엔드 (nginx :8080).
+> 2026-05-22 갱신.
+
+---
+
+## 🟢 현재 상태 (양호)
+
+- 라우트 16개 (12 메인 + 4 서브) 정상 운영
+- agent-office 3×3 그리드 재설계 완료 (5/7~14, WebP 93% 축소, WS 재연결 백오프)
+- `/insta` 슬레이트 캐러셀 + 반응형 (5/15~16)
+- Vite proxy 7개 (NAS API + Fear&Greed + VIX + Treasury + WTI + Brent)
+
+---
+
+## ✅ 최근 완료 (5/18~22)
+
+- 2026-05-22: **거래 데스크 AI 투자 탭 제거** (e42b643) — web-ai signal_v1 legacy 이전과 정합 (V2 단독 운영 반영)
+- 2026-05-22: stock 총 매입을 각 종목 매입가 단순 합으로 표시 (6533743)
+- 2026-05-22: agent-office 모바일 사이드패널 전체화면 토글 + music 에이전트 이미지 교체 (ee5700d)
+- 2026-05-14: agent-office Grid 재설계 (canvas 폐기), AGENT_META + GRID_SLOTS 중앙화
+- 2026-05-15~16: `/insta` 신규 페이지 + InstaCards.jsx + src/api.js(530줄) 음악·인스타·텔레그램 API 확장
+
+---
+
+## 🔴 즉시 (1~3일)
+
+### 1. `/insta` 비동기 폴링 구현 ⭐ (백엔드 준비 완료 → 구현 시점 도래)
+- **배경**: web-backend insta-lab이 Redis 분할(SP-4) 완료 → `_bg_render`가 Redis push, `GET /api/insta/tasks/{task_id}` 폴링 엔드포인트 존재. **이제 frontend가 비동기 폴링으로 전환해야 정합**
+- **파일**: `src/pages/insta/InstaCards.jsx`
+- [ ] 슬레이트 생성 → `task_id` 받고 폴링 (2~5초 간격, NAS 부담 ↓)
+- [ ] progress bar UI (0~100%) + `queue:paused` 상태 표시 (박재오 작업 중 = Windows 워커 정지)
+- [ ] failed 상태 처리 (오류 메시지·재시도 버튼)
+
+### 2. agent-office WebSocket 안정성 점검
+- 5/7~14 재설계 + 5/22 모바일 토글 직후 운영 확인
+- [ ] 브라우저 콘솔 WS 끊김 → 재연결 지수 백오프 실제 작동
+- [ ] 4 테스트(TaskTab·CommandTab·AgentCard·ScoreNodeCard) 통과 재확인
+
+### 3. agent-office lotto sim_consensus 노출
+- **배경**: web-backend `/api/lotto/best`에 5종 점수 array 노출됨 (lotto-signals) + weight-evolver 자율 학습 도입
+- [ ] agent-office lotto 에이전트 카드에 5종 점수·시그널 상태 표시
+- [ ] (선택) weight-evolver 진화 상태 미니 패널
+
+---
+
+## 🟡 중기 (1~2주)
+
+### 4. `/insta` 카드 템플릿 UI 고도화
+- 현재 default theme PNG 미리보기만. hedgy75 테마 추가 시 theme 선택 UI 필요
+- [ ] 테마 선택 dropdown (default / hedgy75)
+- [ ] 미리보기 컴포넌트 페이지 종류별 분기
+
+### 5. `/music` Sonic Forge 발행 모니터링
+- music-lab Redis 분할(SP-6) + Windows music-render 도입 → 발행 상태 모니터링 패널 필요
+- [ ] 발행 큐·실패·재시도 로그 표시 (Redis 큐 길이 연동)
+- [ ] 텔레그램 5단계 승인 UX 점검
+
+### 6. NAS↔Windows 작업 흐름 가시화 (신규)
+- web-ai 워커 3종 + Redis 큐 도입으로 작업 분산 흐름이 복잡해짐
+- [ ] agent-office 또는 신규 `/node` 페이지에 큐 상태·Windows 노드 헬스 표시 (web-ai/web-backend 추가 아이디어와 연동)
+
+---
+
+## 🟢 장기 (1개월+)
+
+### 7. 모바일 UX 일관 적용
+- BottomNav + PullToRefresh + MobileSheet + SwipeableView 있음. 신규 페이지 적용 부족
+- [ ] `/insta` 모바일 캐러셀 swipe + `/agent-office` 모바일 그리드 압축
+
+### 8. `/lab` 페이지 확장
+- 현재 sword-stream · day-calc 2개
+- [ ] 박재오 데모 콘텐츠 큐 결정 (예: weight-evolver 진화 그래프, AI 음악 빠른 청취)
+
+---
+
+## 💡 추가 아이디어 (신규 2026-05-22)
+
+- **`/node` Windows AI 노드 대시보드** — ai_trade + insta/music/video-render + task-watcher 상태, Redis 큐 길이, `queue:paused` 토글 버튼(task-watcher C안 = "토글 UI 1개"). web-ai/web-backend 모니터링 아이디어의 frontend 진입점
+- **video 생성 미리보기 페이지** — video-lab(SP-8) + Windows video-render 4 provider 결과 비교 그리드. 무신사 공모전 MU-진 영상 버전 관리에 활용
+- **weight-evolver 진화 시각화** — auto_picks 적중 추이 + weight base diff 그래프 (`/lab` 또는 lotto 페이지)
+- **위키 페이지 수 정합** — [[사업-개인-웹-플랫폼]]에 "17개" 박혀 있으나 실제 16개 (12 메인 + 4 서브). *박재오 위키 갱신 항목* (web-ui 코드 아님)
+
+---
+
+## 🚀 빌드 & 배포
+
+```bash
+npm run dev          # 개발 (port 3007, Vite proxy)
+npm run build        # 빌드 (rimraf dist + Vite build)
+npm run release:nas  # 자동 배포 (deploy-nas.cjs)
+```
+
+배포: Windows `robocopy dist Z:\docker\webpage\frontend\` / macOS `rsync` → nginx 자동 reload
+
+---
+
+## 📚 참고
+
+- 위키: [[사업-개인-웹-플랫폼]] (백엔드 통합 인덱스)
+- 라우트: `src/routes.jsx` (navLinks 메타) / Vite 프록시: `vite.config.js`
+- API: 모든 페이지 `/api/` 상대 경로 (Mixed Content 방지)
+- 백엔드 짝: web-backend CHECK_POINT (insta-lab Redis 분할 → /insta 비동기 폴링 정합 필요)
+
+## 변경 이력
+
+- 2026-05-18: 페이지 신설. 즉시 3 + 중기 3 + 장기 2.
+- 2026-05-22: 최근 완료 3건 반영(AI 투자 탭 제거·stock 매입 표시·모바일 사이드패널). **`/insta` 비동기 폴링을 즉시 1순위로 승격** (백엔드 insta-lab Redis 분할 완료 → frontend 정합 필요). lotto sim_consensus 노출 + NAS↔Windows 작업 흐름 가시화 항목 추가. 추가 아이디어 4건 신설 (/node 대시보드·video 미리보기·evolver 시각화·위키 페이지 수 정합).

docs/superpowers/plans/2026-05-17-signal-v2-phase4-signal-generator.md

@@ -0,0 +1,817 @@
+# Signal V2 Phase 4 — Signal Generator 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:** signal_v2 에 매수/매도 신호 생성 레이어 추가. Phase 2/3a/3b 의 모든 state 산출 → Phase 0 spec §6.1-§6.3 룰 → `state.signals[ticker]` (Phase 0 spec §5.2 schema) + `SignalDedup` 24h 차단.
+
+**Architecture:** 순수 함수 `generate_signals(state, dedup, settings)` 가 매 분봉 cycle 후 호출. 매수 (Hard gate 4 조건 + soft confidence > 0.7) + 매도 (손절>이상>익절 우선순위). 6 env 외부화 (운영 튜닝).
+
+**Tech Stack:** Python 순수 함수 / pytest / SignalDedup (Phase 2) / 외부 의존성 없음
+
+**Spec:** `web-ui/docs/superpowers/specs/2026-05-17-signal-v2-phase4-signal-generator.md`
+
+---
+
+## 파일 구조
+
+| 파일 | 책임 |
+|------|------|
+| `signal_v2/config.py` | (수정) Settings 에 6 env field 추가 |
+| `signal_v2/state.py` | (수정) PollState `signals` field 추가 |
+| `signal_v2/signal_generator.py` | (신규) `generate_signals(state, dedup, settings)` + 8 helper |
+| `signal_v2/pull_worker.py` | (수정) `poll_loop` signature + 매 cycle 후 `generate_signals` 호출 |
+| `signal_v2/main.py` | (수정) lifespan 의 poll_task 호출에 `dedup` + `settings` 전달 |
+| `signal_v2/tests/test_signal_generator.py` | (신규) 9 단위 케이스 |
+| `signal_v2/tests/test_pull_worker.py` | (수정) integration 1 케이스 추가 |
+
+7 파일 변경, **10 신규 테스트** (45 → 55).
+
+---
+
+## Task 순서
+
+```
+Task 1: foundation (config 6 env + state signals field)
+Task 2: signal_generator.py + 9 단위 tests (TDD)
+Task 3: pull_worker + main.py 통합 + 1 integration test
+Task 4: 사용자 수동 (.env optional + smoke + push)
+```
+
+---
+
+### Task 1: foundation (config + state)
+
+**Files:**
+- Modify: `web-ai/signal_v2/config.py`
+- Modify: `web-ai/signal_v2/state.py`
+
+- [ ] **Step 1: Update config.py with 6 new fields**
+
+Read `web-ai/signal_v2/config.py`. Add 6 fields to Settings (after `chronos_model` field, before properties):
+
+```python
+    stop_loss_pct: float = field(
+        default_factory=lambda: float(os.getenv("STOP_LOSS_PCT", "-0.07"))
+    )
+    take_profit_pct: float = field(
+        default_factory=lambda: float(os.getenv("TAKE_PROFIT_PCT", "0.15"))
+    )
+    chronos_spread_threshold: float = field(
+        default_factory=lambda: float(os.getenv("CHRONOS_SPREAD_THRESHOLD", "0.6"))
+    )
+    asking_bid_ratio_threshold: float = field(
+        default_factory=lambda: float(os.getenv("ASKING_BID_RATIO_THRESHOLD", "0.6"))
+    )
+    confidence_threshold: float = field(
+        default_factory=lambda: float(os.getenv("CONFIDENCE_THRESHOLD", "0.7"))
+    )
+    min_momentum_for_buy: str = field(
+        default_factory=lambda: os.getenv("MIN_MOMENTUM_FOR_BUY", "strong_up")
+    )
+```
+
+- [ ] **Step 2: Update state.py with signals field**
+
+Read `web-ai/signal_v2/state.py`. Add `signals` field to PollState (after `minute_momentum`):
+
+```python
+    signals: dict[str, dict] = field(default_factory=dict)
+```
+
+- [ ] **Step 3: Smoke import test**
+
+```bash
+cd /c/Users/jaeoh/Desktop/workspace/web-ai
+python -c "from signal_v2.config import get_settings; from signal_v2.state import state; s = get_settings(); print(f'stop_loss={s.stop_loss_pct}, conf_threshold={s.confidence_threshold}, min_momentum={s.min_momentum_for_buy}'); print(state)"
+```
+Expected: `stop_loss=-0.07, conf_threshold=0.7, min_momentum=strong_up` + state print with `signals={}`.
+
+- [ ] **Step 4: Run existing tests — no regression**
+
+```bash
+cd /c/Users/jaeoh/Desktop/workspace/web-ai
+python -m pytest signal_v2/tests -q 2>&1 | tail -3
+```
+Expected: 45 passed.
+
+- [ ] **Step 5: Commit**
+
+```bash
+cd /c/Users/jaeoh/Desktop/workspace/web-ai
+git add signal_v2/config.py signal_v2/state.py
+git commit -m "$(cat <<'EOF'
+feat(signal_v2-phase4): foundation — 6 env thresholds + state.signals
+
+config.py: STOP_LOSS_PCT / TAKE_PROFIT_PCT / CHRONOS_SPREAD_THRESHOLD /
+ASKING_BID_RATIO_THRESHOLD / CONFIDENCE_THRESHOLD / MIN_MOMENTUM_FOR_BUY
+env vars with sensible defaults (Phase 0 spec §6.1-§6.2 values).
+
+state.py: PollState.signals dict[ticker, signal_body] for Phase 5 input.
+
+45 existing tests still pass.
+
+Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
+EOF
+)"
+```
+
+---
+
+### Task 2: signal_generator.py + 9 단위 tests
+
+**Files:**
+- Create: `web-ai/signal_v2/signal_generator.py`
+- Create: `web-ai/signal_v2/tests/test_signal_generator.py`
+
+- [ ] **Step 1: Write 9 failing tests**
+
+Create `web-ai/signal_v2/tests/test_signal_generator.py`:
+
+```python
+"""Tests for signal_generator."""
+from collections import deque
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from signal_v2.signal_generator import generate_signals
+from signal_v2.state import PollState
+
+
+def _settings(**overrides):
+    """Build a Settings-like object for tests (avoid env)."""
+    defaults = dict(
+        stop_loss_pct=-0.07,
+        take_profit_pct=0.15,
+        chronos_spread_threshold=0.6,
+        asking_bid_ratio_threshold=0.6,
+        confidence_threshold=0.7,
+        min_momentum_for_buy="strong_up",
+    )
+    defaults.update(overrides)
+    m = MagicMock()
+    for k, v in defaults.items():
+        setattr(m, k, v)
+    return m
+
+
+def _make_state_with_buy_candidate(
+    ticker="005930", name="삼성전자", rank=1,
+    chronos_median=0.02, chronos_q10=-0.01, chronos_q90=0.04, chronos_conf=0.85,
+    momentum="strong_up", bid_ratio=0.7, current_price=78500,
+):
+    state = PollState()
+    state.screener_preview = {"items": [{"ticker": ticker, "name": name}]}
+    state.chronos_predictions[ticker] = {
+        "median": chronos_median, "q10": chronos_q10, "q90": chronos_q90,
+        "conf": chronos_conf, "as_of": "2026-05-17T16:00:00+09:00",
+    }
+    state.minute_momentum[ticker] = momentum
+    state.asking_price[ticker] = {
+        "bid_total": int(bid_ratio * 1000),
+        "ask_total": int((1 - bid_ratio) * 1000),
+        "bid_ratio": bid_ratio,
+        "current_price": current_price,
+        "as_of": "2026-05-17T16:00:01+09:00",
+    }
+    return state
+
+
+def _make_state_with_holding(
+    ticker="005930", name="삼성전자",
+    pnl_pct=0.0, avg_price=75000, current_price=75000,
+):
+    state = PollState()
+    state.portfolio = {"holdings": [{
+        "ticker": ticker, "name": name,
+        "avg_price": avg_price, "current_price": current_price,
+        "pnl_pct": pnl_pct, "profit_rate": pnl_pct * 100,
+        "quantity": 100, "broker": "키움",
+    }]}
+    state.screener_preview = {"items": []}
+    return state
+
+
+@pytest.fixture
+def dedup_mock():
+    d = MagicMock()
+    d.is_recent.return_value = False
+    return d
+
+
+def test_buy_signal_when_all_conditions_pass_and_confidence_high(dedup_mock):
+    state = _make_state_with_buy_candidate()
+    generate_signals(state, dedup_mock, _settings())
+    assert "005930" in state.signals
+    sig = state.signals["005930"]
+    assert sig["action"] == "buy"
+    assert sig["confidence_webai"] > 0.7
+    dedup_mock.record.assert_called()
+
+
+def test_silent_when_chronos_median_negative(dedup_mock):
+    state = _make_state_with_buy_candidate(chronos_median=-0.01)
+    generate_signals(state, dedup_mock, _settings())
+    assert "005930" not in state.signals
+
+
+def test_silent_when_distribution_spread_too_wide(dedup_mock):
+    # spread = (0.5 - (-0.5)) / max(|0.001|, 0.001) = 1000 → > 0.6
+    state = _make_state_with_buy_candidate(
+        chronos_median=0.001, chronos_q10=-0.5, chronos_q90=0.5,
+    )
+    generate_signals(state, dedup_mock, _settings())
+    assert "005930" not in state.signals
+
+
+def test_silent_when_momentum_not_strong_up(dedup_mock):
+    state = _make_state_with_buy_candidate(momentum="weak_up")
+    generate_signals(state, dedup_mock, _settings())
+    assert "005930" not in state.signals
+
+
+def test_silent_when_bid_ratio_below_threshold(dedup_mock):
+    state = _make_state_with_buy_candidate(bid_ratio=0.5)
+    generate_signals(state, dedup_mock, _settings())
+    assert "005930" not in state.signals
+
+
+def test_silent_when_confidence_below_threshold(dedup_mock):
+    # chronos_conf low + rank=20 → confidence < 0.7
+    state = _make_state_with_buy_candidate(chronos_conf=0.3)
+    # add 19 fake items to push rank to 20
+    state.screener_preview["items"] = (
+        [{"ticker": f"FAKE{i:03d}"} for i in range(19)]
+        + [{"ticker": "005930", "name": "삼성전자"}]
+    )
+    generate_signals(state, dedup_mock, _settings())
+    # confidence_webai = 0.3*0.5 + 1.0*0.3 + 0.05*0.2 = 0.15 + 0.3 + 0.01 = 0.46 < 0.7
+    assert "005930" not in state.signals
+
+
+def test_sell_signal_when_stop_loss_triggered(dedup_mock):
+    state = _make_state_with_holding(pnl_pct=-0.08, current_price=69000, avg_price=75000)
+    generate_signals(state, dedup_mock, _settings())
+    assert "005930" in state.signals
+    sig = state.signals["005930"]
+    assert sig["action"] == "sell"
+    assert sig["confidence_webai"] == 1.0  # 손절선 즉시
+    assert sig["pnl_pct"] == -0.08
+
+
+def test_sell_signal_when_take_profit_triggered(dedup_mock):
+    state = _make_state_with_holding(pnl_pct=0.16, current_price=87000, avg_price=75000)
+    generate_signals(state, dedup_mock, _settings())
+    assert "005930" in state.signals
+    sig = state.signals["005930"]
+    assert sig["action"] == "sell"
+    assert sig["confidence_webai"] == 0.6  # 익절선 검토 알림
+
+
+def test_silent_when_dedup_recently_sent(dedup_mock):
+    state = _make_state_with_buy_candidate()
+    dedup_mock.is_recent.return_value = True  # dedup 차단
+    generate_signals(state, dedup_mock, _settings())
+    assert "005930" not in state.signals
+    dedup_mock.record.assert_not_called()
+```
+
+- [ ] **Step 2: Run tests to verify FAIL**
+
+```bash
+cd /c/Users/jaeoh/Desktop/workspace/web-ai
+python -m pytest signal_v2/tests/test_signal_generator.py -v 2>&1 | tail -10
+```
+Expected: ImportError (signal_v2.signal_generator missing).
+
+- [ ] **Step 3: Implement signal_generator.py**
+
+Create `web-ai/signal_v2/signal_generator.py`:
+
+```python
+"""Phase 4 — 매수/매도 신호 생성.
+
+순수 함수 generate_signals(state, dedup, settings). state 를 mutate.
+"""
+from __future__ import annotations
+import logging
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+logger = logging.getLogger(__name__)
+KST = ZoneInfo("Asia/Seoul")
+
+# 분봉 모멘텀 → linear score
+MOMENTUM_SCORES = {
+    "strong_up": 1.0,
+    "weak_up": 0.7,
+    "neutral": 0.5,
+    "weak_down": 0.3,
+    "strong_down": 0.0,
+}
+
+
+def generate_signals(state, dedup, settings) -> None:
+    """Phase 4 entry — state mutating. 매수/매도 룰 적용."""
+    _evaluate_buy_signals(state, dedup, settings)
+    _evaluate_sell_signals(state, dedup, settings)
+
+
+# ----- 매수 -----
+
+def _evaluate_buy_signals(state, dedup, settings) -> None:
+    candidates = _buy_candidates(state)
+    for ticker, name, rank in candidates:
+        if not _check_buy_hard_gate(state, ticker, settings):
+            continue
+        confidence = _compute_buy_confidence(state, ticker, rank)
+        if confidence <= settings.confidence_threshold:
+            continue
+        if dedup.is_recent(ticker, "buy", within_hours=24):
+            continue
+        state.signals[ticker] = _build_buy_signal(state, ticker, name, rank, confidence)
+        dedup.record(ticker, "buy", confidence=confidence)
+
+
+def _buy_candidates(state) -> list[tuple[str, str, int | None]]:
+    """screener Top-N (rank 1..N) + portfolio (rank=None)."""
+    candidates: list[tuple[str, str, int | None]] = []
+    seen: set[str] = set()
+    # Screener Top-N
+    if state.screener_preview is not None:
+        for i, item in enumerate(state.screener_preview.get("items", [])):
+            ticker = item.get("ticker")
+            if not ticker or ticker in seen:
+                continue
+            seen.add(ticker)
+            name = item.get("name", ticker)
+            candidates.append((ticker, name, i + 1))
+    # Portfolio holdings
+    if state.portfolio is not None:
+        for h in state.portfolio.get("holdings", []):
+            ticker = h.get("ticker")
+            if not ticker or ticker in seen:
+                continue
+            seen.add(ticker)
+            candidates.append((ticker, h.get("name", ticker), None))
+    return candidates
+
+
+def _check_buy_hard_gate(state, ticker: str, settings) -> bool:
+    pred = state.chronos_predictions.get(ticker)
+    if pred is None or pred["median"] <= 0:
+        return False
+    spread = (pred["q90"] - pred["q10"]) / max(abs(pred["median"]), 0.001)
+    if spread >= settings.chronos_spread_threshold:
+        return False
+    momentum = state.minute_momentum.get(ticker)
+    if momentum != settings.min_momentum_for_buy:
+        return False
+    ap = state.asking_price.get(ticker)
+    if ap is None or ap["bid_ratio"] < settings.asking_bid_ratio_threshold:
+        return False
+    return True
+
+
+def _compute_buy_confidence(state, ticker: str, rank: int | None) -> float:
+    pred = state.chronos_predictions[ticker]
+    chronos_conf = pred["conf"]
+    minute_score = MOMENTUM_SCORES.get(state.minute_momentum.get(ticker, "neutral"), 0.5)
+    screener_norm = 1 - (rank - 1) / 20 if rank is not None else 0.0
+    return chronos_conf * 0.5 + minute_score * 0.3 + screener_norm * 0.2
+
+
+def _build_buy_signal(state, ticker: str, name: str, rank: int | None, confidence: float) -> dict:
+    ap = state.asking_price[ticker]
+    pred = state.chronos_predictions[ticker]
+    return {
+        "ticker": ticker,
+        "name": name,
+        "action": "buy",
+        "confidence_webai": confidence,
+        "current_price": ap["current_price"],
+        "avg_price": None,
+        "pnl_pct": None,
+        "context": _build_context(state, ticker, rank),
+        "as_of": datetime.now(KST).isoformat(),
+    }
+
+
+# ----- 매도 -----
+
+def _evaluate_sell_signals(state, dedup, settings) -> None:
+    if state.portfolio is None:
+        return
+    for holding in state.portfolio.get("holdings", []):
+        ticker = holding.get("ticker")
+        if not ticker:
+            continue
+        sell = _try_stop_loss(state, holding, settings)
+        if sell is None:
+            sell = _try_anomaly(state, holding, settings)
+        if sell is None:
+            sell = _try_take_profit(state, holding, settings)
+        if sell is None:
+            continue
+        if dedup.is_recent(ticker, "sell", within_hours=24):
+            continue
+        state.signals[ticker] = sell
+        dedup.record(ticker, "sell", confidence=sell["confidence_webai"])
+
+
+def _try_stop_loss(state, holding: dict, settings) -> dict | None:
+    pnl = holding.get("pnl_pct")
+    if pnl is None or pnl >= settings.stop_loss_pct:
+        return None
+    return _build_sell_signal(state, holding, confidence=1.0, reason="stop_loss")
+
+
+def _try_take_profit(state, holding: dict, settings) -> dict | None:
+    pnl = holding.get("pnl_pct")
+    if pnl is None or pnl <= settings.take_profit_pct:
+        return None
+    return _build_sell_signal(state, holding, confidence=0.6, reason="take_profit")
+
+
+def _try_anomaly(state, holding: dict, settings) -> dict | None:
+    ticker = holding["ticker"]
+    pred = state.chronos_predictions.get(ticker)
+    if pred is None or pred["median"] >= -0.01:
+        return None
+    momentum = state.minute_momentum.get(ticker)
+    if momentum != "strong_down":
+        return None
+    ap = state.asking_price.get(ticker)
+    if ap is None:
+        return None
+    if ap["bid_ratio"] > (1 - settings.asking_bid_ratio_threshold):
+        return None  # 매도세 60% 미만
+    minute_score = 1.0 - MOMENTUM_SCORES.get(momentum, 0.5)  # 반전
+    confidence = pred["conf"] * 0.5 + minute_score * 0.3 + 1.0 * 0.2
+    if confidence <= settings.confidence_threshold:
+        return None
+    return _build_sell_signal(state, holding, confidence=confidence, reason="anomaly")
+
+
+def _build_sell_signal(state, holding: dict, confidence: float, reason: str) -> dict:
+    ticker = holding["ticker"]
+    return {
+        "ticker": ticker,
+        "name": holding.get("name", ticker),
+        "action": "sell",
+        "confidence_webai": confidence,
+        "current_price": holding.get("current_price"),
+        "avg_price": holding.get("avg_price"),
+        "pnl_pct": holding.get("pnl_pct"),
+        "context": _build_context(state, ticker, rank=None, sell_reason=reason),
+        "as_of": datetime.now(KST).isoformat(),
+    }
+
+
+# ----- Context -----
+
+def _build_context(state, ticker: str, rank: int | None, sell_reason: str | None = None) -> dict:
+    pred = state.chronos_predictions.get(ticker) or {}
+    ap = state.asking_price.get(ticker) or {}
+    news_item = _find_news_sentiment(state, ticker)
+    screener_scores = _find_screener_scores(state, ticker)
+    context: dict = {
+        "chronos_pred_1d": pred.get("median"),
+        "chronos_pred_conf": pred.get("conf"),
+        "chronos_q10": pred.get("q10"),
+        "chronos_q90": pred.get("q90"),
+        "screener_rank": rank,
+        "screener_scores": screener_scores,
+        "minute_momentum": state.minute_momentum.get(ticker),
+        "asking_bid_ratio": ap.get("bid_ratio"),
+        "news_sentiment": news_item.get("score") if news_item else None,
+        "news_reason": news_item.get("reason") if news_item else None,
+    }
+    if sell_reason is not None:
+        context["sell_reason"] = sell_reason
+    return context
+
+
+def _find_news_sentiment(state, ticker: str) -> dict | None:
+    if state.news_sentiment is None:
+        return None
+    for item in state.news_sentiment.get("items", []):
+        if item.get("ticker") == ticker:
+            return item
+    return None
+
+
+def _find_screener_scores(state, ticker: str) -> dict | None:
+    if state.screener_preview is None:
+        return None
+    for item in state.screener_preview.get("items", []):
+        if item.get("ticker") == ticker:
+            return item.get("scores")
+    return None
+```
+
+- [ ] **Step 4: Run tests to verify PASS**
+
+```bash
+cd /c/Users/jaeoh/Desktop/workspace/web-ai
+python -m pytest signal_v2/tests/test_signal_generator.py -v 2>&1 | tail -15
+```
+Expected: 9 passed.
+
+Full suite:
+```bash
+python -m pytest signal_v2/tests -q 2>&1 | tail -3
+```
+Expected: 54 passed.
+
+If any test fails, examine the assertion + impl. Common gotchas:
+- Confidence calculation order — chronos*0.5 + minute*0.3 + screener*0.2
+- Stop loss `<` (strict) vs `<=` — spec says "도달 시" so use `<` strict
+- screener_norm when rank=None → 0.0 (not 1.0)
+
+- [ ] **Step 5: Commit**
+
+```bash
+cd /c/Users/jaeoh/Desktop/workspace/web-ai
+git add signal_v2/signal_generator.py signal_v2/tests/test_signal_generator.py
+git commit -m "$(cat <<'EOF'
+feat(signal_v2-phase4): signal_generator + 9 unit tests
+
+generate_signals(state, dedup, settings) → state mutating:
+- Buy: screener Top-N + portfolio. Hard gate (chronos median > 0 +
+  spread < 0.6 + momentum strong_up + bid_ratio >= 0.6) + soft
+  confidence (chronos*0.5 + minute*0.3 + screener*0.2) > 0.7.
+- Sell: portfolio only. Priority stop_loss > anomaly > take_profit.
+  Stop loss confidence 1.0 (immediate), take_profit 0.6 (review).
+- SignalDedup 24h via dedup.is_recent/record per (ticker, action).
+- State signal dict matches Phase 0 spec §5.2 schema.
+
+54 tests pass.
+
+Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
+EOF
+)"
+```
+
+---
+
+### Task 3: pull_worker + main.py integration + 1 test
+
+**Files:**
+- Modify: `web-ai/signal_v2/pull_worker.py`
+- Modify: `web-ai/signal_v2/main.py`
+- Modify: `web-ai/signal_v2/tests/test_pull_worker.py`
+
+- [ ] **Step 1: Write failing integration test**
+
+Append to `web-ai/signal_v2/tests/test_pull_worker.py`:
+
+```python
+def test_poll_loop_calls_generate_signals_after_cycle(monkeypatch):
+    """매 cycle 후 generate_signals 호출 + state.signals 갱신."""
+    from unittest.mock import MagicMock
+    from signal_v2.state import PollState
+
+    state = PollState()
+    state.portfolio = {"holdings": [{
+        "ticker": "005930", "name": "삼성전자",
+        "avg_price": 75000, "current_price": 69000,
+        "pnl_pct": -0.08, "profit_rate": -8.0,
+        "quantity": 100, "broker": "키움",
+    }]}
+    state.screener_preview = {"items": []}
+
+    dedup = MagicMock()
+    dedup.is_recent.return_value = False
+
+    settings = MagicMock()
+    settings.stop_loss_pct = -0.07
+    settings.take_profit_pct = 0.15
+    settings.chronos_spread_threshold = 0.6
+    settings.asking_bid_ratio_threshold = 0.6
+    settings.confidence_threshold = 0.7
+    settings.min_momentum_for_buy = "strong_up"
+
+    from signal_v2.signal_generator import generate_signals
+    # Call generate_signals directly to verify state mutation through the public API.
+    generate_signals(state, dedup, settings)
+
+    # Stop loss should trigger
+    assert "005930" in state.signals
+    assert state.signals["005930"]["action"] == "sell"
+    assert state.signals["005930"]["confidence_webai"] == 1.0
+    dedup.record.assert_called_with("005930", "sell", confidence=1.0)
+```
+
+- [ ] **Step 2: Run test to verify PASS (signal_generator from Task 2 already exists)**
+
+```bash
+cd /c/Users/jaeoh/Desktop/workspace/web-ai
+python -m pytest signal_v2/tests/test_pull_worker.py::test_poll_loop_calls_generate_signals_after_cycle -v 2>&1 | tail -10
+```
+Expected: PASS (test exercises generate_signals directly — public API integration).
+
+- [ ] **Step 3: Update pull_worker.py — poll_loop signature + cycle integration**
+
+Read `web-ai/signal_v2/pull_worker.py`. Modify the `poll_loop` signature to accept dedup + settings:
+
+```python
+async def poll_loop(
+    client, state, shutdown,
+    kis_client=None, chronos=None,
+    dedup=None, settings=None,
+) -> None:
+    """...existing docstring..."""
+    logger.info("poll_loop started")
+    while not shutdown.is_set():
+        now = datetime.now(KST)
+        if _is_market_day(now) and _is_polling_window(now):
+            try:
+                await _run_polling_cycle(client, state, kis_client=kis_client)
+            except Exception:
+                logger.exception("poll cycle failed")
+            try:
+                update_minute_momentum_for_all(state)
+            except Exception:
+                logger.exception("minute momentum update failed")
+            if _is_post_close_trigger(now) and chronos is not None and kis_client is not None:
+                try:
+                    await _run_post_close_cycle(kis_client, chronos, state)
+                except Exception:
+                    logger.exception("post-close cycle failed")
+            # Phase 4: generate signals
+            if dedup is not None and settings is not None:
+                try:
+                    from signal_v2.signal_generator import generate_signals
+                    generate_signals(state, dedup, settings)
+                except Exception:
+                    logger.exception("generate_signals failed")
+        interval = _next_interval(now)
+        try:
+            await asyncio.wait_for(shutdown.wait(), timeout=interval)
+            break
+        except asyncio.TimeoutError:
+            continue
+    logger.info("poll_loop ended")
+```
+
+- [ ] **Step 4: Update main.py — pass dedup + settings to poll_loop**
+
+Read `web-ai/signal_v2/main.py`. Find the `asyncio.create_task(poll_loop(...))` call inside `lifespan` and add `dedup` + `settings` params:
+
+```python
+    _ctx.poll_task = asyncio.create_task(
+        poll_loop(
+            _ctx.client, state_mod.state, _ctx.shutdown,
+            kis_client=_ctx.kis_client,
+            chronos=_ctx.chronos,
+            dedup=_ctx.dedup,
+            settings=settings,
+        )
+    )
+```
+
+- [ ] **Step 5: Run full test suite**
+
+```bash
+cd /c/Users/jaeoh/Desktop/workspace/web-ai
+python -m pytest signal_v2/tests -q 2>&1 | tail -3
+```
+Expected: 55 passed (54 + 1 new integration).
+
+- [ ] **Step 6: Commit**
+
+```bash
+cd /c/Users/jaeoh/Desktop/workspace/web-ai
+git add signal_v2/pull_worker.py signal_v2/main.py signal_v2/tests/test_pull_worker.py
+git commit -m "$(cat <<'EOF'
+feat(signal_v2-phase4): pull_worker + main.py integrate signal generator
+
+poll_loop signature now accepts dedup + settings. After each cycle
+(stock pull + minute momentum + post-close), call generate_signals
+to populate state.signals. main.py lifespan passes _ctx.dedup and
+settings to poll_loop.
+
+1 integration test added. 55 tests pass.
+
+Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
+EOF
+)"
+```
+
+---
+
+### Task 4: 사용자 수동 — .env optional + smoke + push
+
+**This task requires user action.**
+
+- [ ] **Step 1: .env optional**
+
+6 env 의 default 가 Phase 0 spec 값과 동일 — `.env` 변경 불필요. 운영 검증 후 조정 시:
+```
+STOP_LOSS_PCT=-0.07
+TAKE_PROFIT_PCT=0.15
+CHRONOS_SPREAD_THRESHOLD=0.6
+ASKING_BID_RATIO_THRESHOLD=0.6
+CONFIDENCE_THRESHOLD=0.7
+MIN_MOMENTUM_FOR_BUY=strong_up
+```
+
+- [ ] **Step 2: signal_v2 재시작**
+
+기존 signal_v2 가 가동 중이면 Ctrl+C 후:
+```powershell
+cd C:\Users\jaeoh\Desktop\workspace\web-ai\signal_v2
+.\start.bat
+```
+기대: 정상 시작 (signal_generator 자동 호출 — 매 cycle 마다).
+
+- [ ] **Step 3: state.signals 검증 (수동)**
+
+운영 시간대라면 cycle 진행 + state.signals 채워질 수 있음. 수동 검증:
+```powershell
+cd C:\Users\jaeoh\Desktop\workspace\web-ai
+python -c "
+import asyncio
+from signal_v2.config import get_settings
+from signal_v2.kis_client import KISClient
+from signal_v2.chronos_predictor import ChronosPredictor
+from signal_v2.state import PollState
+from signal_v2.rate_limit import SignalDedup
+from signal_v2.pull_worker import _run_post_close_cycle, update_minute_momentum_for_all
+from signal_v2.signal_generator import generate_signals
+
+async def main():
+    s = get_settings()
+    kc = KISClient(app_key=s.kis_app_key, app_secret=s.kis_app_secret, account=s.kis_account, is_virtual=s.kis_is_virtual, v1_token_path=s.v1_token_path)
+    cp = ChronosPredictor(model_name=s.chronos_model)
+    dedup = SignalDedup(s.db_path)
+    state = PollState()
+    state.portfolio = {'holdings': [{'ticker': '005930', 'name': '삼성전자', 'avg_price': 75000, 'current_price': 78500, 'pnl_pct': 0.047, 'profit_rate': 4.67, 'quantity': 100, 'broker': '키움'}]}
+    state.screener_preview = {'items': []}
+    try:
+        await _run_post_close_cycle(kc, cp, state)
+        update_minute_momentum_for_all(state)
+        generate_signals(state, dedup, s)
+        print('Signals:', state.signals)
+    finally:
+        await kc.close()
+asyncio.run(main())
+"
+```
+Expected: `Signals: {}` (정상 — pnl_pct 0.047 은 손절/익절 트리거 안 함, 매수 조건 다 만족 어려움) 또는 일부 신호 dict.
+
+- [ ] **Step 4: V1 무영향**
+
+V1 정상 가동 확인.
+
+- [ ] **Step 5: push**
+
+```powershell
+cd C:\Users\jaeoh\Desktop\workspace\web-ai
+git push
+```
+
+- [ ] **Step 6: 결과 보고**
+
+- Step 2 (signal_v2 시작): PASS / FAIL
+- Step 3 (state.signals 검증): PASS — empty dict or 신호 결과 공유 / FAIL
+- Step 4 (V1 무영향): PASS / FAIL
+- Step 5 (push): PASS / FAIL
+
+전체 PASS 시 **Phase 4 완료** → Phase 5 (agent-office /signal + Qwen3 + 이중 텔레그램) brainstorming.
+
+---
+
+## Self-Review
+
+**1. Spec coverage:**
+
+| Spec § | 요구사항 | Plan task |
+|--------|----------|----------|
+| §2 ① signal_generator | Task 2 ✅ |
+| §2 ② config 6 env | Task 1 ✅ |
+| §2 ③ state.signals | Task 1 ✅ |
+| §2 ④ pull_worker integration | Task 3 ✅ |
+| §2 ⑤ main.py lifespan | Task 3 ✅ |
+| §2 ⑥ 10 tests | Task 2 (9) + Task 3 (1) = 10 ✅ |
+| §4 매수 룰 + confidence | Task 2 (_check_buy_hard_gate + _compute_buy_confidence) ✅ |
+| §5 매도 룰 + dedup | Task 2 (_try_stop_loss/anomaly/take_profit + dedup.is_recent/record) ✅ |
+| §6 state 통합 + pull_worker | Task 1 + Task 3 ✅ |
+| §7 signal_generator 구조 | Task 2 Step 3 (8 helpers) ✅ |
+| §8 10 테스트 케이스 | Task 2-3 ✅ |
+| §9 DoD 8 항목 | Task 1-4 합산 ✅ |
+
+No gaps.
+
+**2. Placeholder scan**: No "TBD" / "implement later". 각 step 의 코드 + 명령 모두 명시.
+
+**3. Type consistency:**
+- `generate_signals(state, dedup, settings) -> None` consistent Task 2 + Task 3 ✅
+- `MOMENTUM_SCORES` 매핑 consistent (1.0/0.7/0.5/0.3/0.0) ✅
+- Settings field names consistent Task 1 + Task 2 (stop_loss_pct, etc.) ✅
+- PollState.signals dict[str, dict] consistent ✅
+- helper signatures (_check_buy_hard_gate, _compute_buy_confidence, _try_stop_loss, _try_anomaly, _try_take_profit, _build_buy_signal, _build_sell_signal, _build_context) consistent ✅
+
+Plan passes self-review.

docs/superpowers/plans/2026-05-25-ai-trade-hotfix.md

@@ -0,0 +1,642 @@
+# ai_trade Hotfix — Code Review F1·F2·F3·F4 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:** ai_trade(V2) 코드 리뷰 7개 finding 중 High 3건(F1·F2·F3) + Medium 1건(F4)을 TDD로 수정. F5/F6은 별도 plan, F7은 pushback.
+
+**Architecture:** 모두 ai_trade/ 내부 단일 모듈 수정. (1) config.py default 경로 — legacy/ 경유. (2) kis_client.py — asyncio.Lock으로 `_throttle()` 직렬화. (3) scheduler.py + pull_worker.py — post-close를 시간 윈도우가 아닌 "일 1회 + 16:00 이후" 상태기반으로 변경. (4) chronos_predictor.py — confidence 산식을 absolute spread 기반으로 통일.
+
+**Tech Stack:** Python 3.12, asyncio, pytest + pytest-asyncio + respx, httpx.
+
+**Test runner:** `.venv` 한글 경로 깨짐 + 리뷰어 Python 312 경로 부재 보고로, 시스템 Python 사용. 정확한 경로는 `where python` 으로 우선 확인. 기본 시도 순서:
+1. `python -m pytest ai_trade/tests -q` (PATH의 Python)
+2. `py -3.12 -m pytest ai_trade/tests -q` (py launcher)
+3. 둘 다 실패 시 환경 셋업이 선행 작업 — plan 진행 중단하고 박재오에게 보고.
+
+**Working directory:** `C:\Users\jaeoh\Desktop\workspace\web-ai` (web-ai repo). Commit/push도 이 디렉토리에서만.
+
+---
+
+## File Map
+
+| 파일 | 변경 종류 | 책임 |
+|------|-----------|------|
+| `ai_trade/config.py` | Modify L31-36 | V1_TOKEN_PATH default를 `legacy/signal_v1/data/kis_token.json`로 |
+| `ai_trade/kis_client.py` | Modify L40-62 | `_throttle_lock: asyncio.Lock` 추가, `_throttle()`을 lock 안에서 실행 |
+| `ai_trade/scheduler.py` | Modify L79-84 | `_is_post_close_trigger(now, last_post_close_date)` 시그니처 변경 — 상태기반 |
+| `ai_trade/pull_worker.py` | Modify L1-58 | `poll_loop`에 `last_post_close_date` state 추가, 호출부 갱신 |
+| `ai_trade/chronos_predictor.py` | Modify L106, L127 | spread 계산을 absolute (q90-q10)로, confidence 산식 `max(0, 1 - spread/0.6)` |
+| `ai_trade/tests/test_kis_client.py` | Add 1 test | concurrent gather throttle test |
+| `ai_trade/tests/test_scheduler.py` | Add 3 tests | post-close 상태기반 트리거 |
+| `ai_trade/tests/test_pull_worker.py` | Add 1 test | 첫 호출 안 됐다가 16:00 이후 5분 cycle에서 호출됨 |
+| `ai_trade/tests/test_chronos_predictor.py` | Add 2 tests | median≈0에서도 conf 정상, spread 클수록 conf↓ |
+| `ai_trade/tests/test_main.py` | Modify | v1_token_path default 변경 반영 (있다면) |
+
+---
+
+## Task 1: F1 — KIS 토큰 경로 default를 legacy/ 경유로
+
+**Files:**
+- Modify: `ai_trade/config.py:31-36`
+- Test: `ai_trade/tests/test_config_token_path.py` (Create)
+
+- [ ] **Step 1: Write the failing test**
+
+```python
+# ai_trade/tests/test_config_token_path.py
+"""F1 — V1_TOKEN_PATH default가 legacy/signal_v1/ 경유인지 검증."""
+import os
+from pathlib import Path
+
+from ai_trade.config import Settings
+
+
+def test_v1_token_default_path_uses_legacy_dir(monkeypatch):
+    """env에 V1_TOKEN_PATH 없으면 legacy/signal_v1/data/kis_token.json"""
+    monkeypatch.delenv("V1_TOKEN_PATH", raising=False)
+    settings = Settings()
+    expected_suffix = Path("legacy") / "signal_v1" / "data" / "kis_token.json"
+    assert str(settings.v1_token_path).endswith(str(expected_suffix)), (
+        f"expected default to end with {expected_suffix}, got {settings.v1_token_path}"
+    )
+
+
+def test_v1_token_env_override_wins(monkeypatch, tmp_path):
+    """env로 명시한 경로가 default를 덮어씀."""
+    custom = tmp_path / "custom_token.json"
+    monkeypatch.setenv("V1_TOKEN_PATH", str(custom))
+    settings = Settings()
+    assert settings.v1_token_path == custom
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+```
+python -m pytest ai_trade/tests/test_config_token_path.py -v
+```
+
+Expected: `test_v1_token_default_path_uses_legacy_dir` FAILs (default가 `signal_v1/...` 임). env override는 PASS.
+
+- [ ] **Step 3: Fix config.py**
+
+`ai_trade/config.py:31-36` 변경:
+
+```python
+    v1_token_path: Path = field(
+        default_factory=lambda: Path(
+            os.getenv("V1_TOKEN_PATH",
+                      str(Path(__file__).parent.parent / "legacy" / "signal_v1" / "data" / "kis_token.json"))
+        )
+    )
+```
+
+- [ ] **Step 4: Run test to verify it passes**
+
+```
+python -m pytest ai_trade/tests/test_config_token_path.py -v
+```
+
+Expected: 2 passed.
+
+- [ ] **Step 5: Verify full test suite still passes**
+
+```
+python -m pytest ai_trade/tests -q
+```
+
+Expected: 모든 기존 테스트 PASS (token path 기본값 변경이 다른 test에 영향 없는지 확인).
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add ai_trade/config.py ai_trade/tests/test_config_token_path.py
+git commit -m "fix(ai_trade): V1_TOKEN_PATH default를 legacy/signal_v1/ 경유로 수정 (F1)"
+```
+
+---
+
+## Task 2: F2 — KIS throttle을 asyncio.Lock으로 직렬화
+
+**Files:**
+- Modify: `ai_trade/kis_client.py:40-62`
+- Test: `ai_trade/tests/test_kis_client.py` (Modify — 새 test 추가)
+
+- [ ] **Step 1: Write the failing test**
+
+`ai_trade/tests/test_kis_client.py` 파일 끝에 추가:
+
+```python
+import asyncio
+import time as time_module
+
+
+@respx.mock
+async def test_throttle_serializes_concurrent_gather(kis_client_factory):
+    """5개 동시 요청이 asyncio.gather로 들어와도 0.5초 간격으로 직렬화되어야 함 (F2).
+
+    초당 2회 = 0.5초 간격. 5개 요청이면 최소 (5-1)*0.5 = 2.0초 소요.
+    Race condition 있으면 5개가 거의 동시에 나가서 2초 훨씬 안쪽에 끝남.
+    """
+    sample = {"output2": []}
+    respx.get(
+        "https://openapivts.koreainvestment.com:29443/uapi/domestic-stock/v1/quotations/inquire-time-itemchartprice"
+    ).mock(return_value=httpx.Response(200, json=sample))
+
+    client = kis_client_factory()
+    try:
+        start = time_module.monotonic()
+        await asyncio.gather(*[client.get_minute_ohlcv(f"00593{i}") for i in range(5)])
+        elapsed = time_module.monotonic() - start
+        # 5개 throttle = 최소 (5-1)*0.5 = 2.0초. tolerance 0.3초.
+        assert elapsed >= 1.7, (
+            f"throttle race condition: 5 concurrent calls took only {elapsed:.2f}s, "
+            f"expected >=1.7s (0.5s * 4 inter-call gaps)"
+        )
+    finally:
+        await client.close()
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+```
+python -m pytest ai_trade/tests/test_kis_client.py::test_throttle_serializes_concurrent_gather -v
+```
+
+Expected: FAIL — elapsed가 0.5초 이하 (race로 동시 깸).
+
+- [ ] **Step 3: Add asyncio.Lock to KISClient**
+
+`ai_trade/kis_client.py:40` `__init__` 끝에 한 줄 추가:
+
+```python
+        self._token_cache: tuple[str, float] | None = None  # (token, file_mtime)
+        self._last_throttle_at = 0.0
+        self._throttle_lock = asyncio.Lock()
+```
+
+그리고 `_throttle()` (L58-62)을 lock으로 감쌈:
+
+```python
+    async def _throttle(self) -> None:
+        async with self._throttle_lock:
+            elapsed = time.monotonic() - self._last_throttle_at
+            if elapsed < _THROTTLE_INTERVAL:
+                await asyncio.sleep(_THROTTLE_INTERVAL - elapsed)
+            self._last_throttle_at = time.monotonic()
+```
+
+- [ ] **Step 4: Run test to verify it passes**
+
+```
+python -m pytest ai_trade/tests/test_kis_client.py::test_throttle_serializes_concurrent_gather -v
+```
+
+Expected: PASS — elapsed >= 1.7s.
+
+- [ ] **Step 5: Verify full kis_client suite still passes**
+
+```
+python -m pytest ai_trade/tests/test_kis_client.py -v
+```
+
+Expected: 모든 test PASS (기존 429 retry 등 영향 없는지 확인).
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add ai_trade/kis_client.py ai_trade/tests/test_kis_client.py
+git commit -m "fix(ai_trade): KIS throttle을 asyncio.Lock으로 직렬화 (F2)"
+```
+
+---
+
+## Task 3: F3 — post-close 트리거를 상태기반으로 변경
+
+**Files:**
+- Modify: `ai_trade/scheduler.py:79-84`
+- Modify: `ai_trade/pull_worker.py:1-58`
+- Test: `ai_trade/tests/test_scheduler.py` (add 3 tests)
+
+**Why state-based:** 16:00:00-16:00:59 윈도우는 5분 sleep + 비결정적 cycle 시작 시각과 충돌. "오늘 아직 post-close 안 돌렸고 현재 시각 ≥ 16:00 이면 trigger 후 today 표시" 로 변경.
+
+- [ ] **Step 1: Write the failing tests**
+
+`ai_trade/tests/test_scheduler.py` 파일 끝에 추가:
+
+```python
+from datetime import date as _date
+from ai_trade.scheduler import _is_post_close_trigger
+
+
+def test_post_close_trigger_fires_at_1601_if_not_yet_today():
+    """16:01에 깬 cycle도 오늘 아직 안 돌렸으면 trigger (F3)."""
+    now = _kst(2026, 5, 18, 16, 1)
+    assert _is_post_close_trigger(now, last_post_close_date=None) is True
+
+
+def test_post_close_trigger_skips_if_already_today():
+    """이미 오늘 돌렸으면 trigger 안 함."""
+    now = _kst(2026, 5, 18, 16, 5)
+    today = _date(2026, 5, 18)
+    assert _is_post_close_trigger(now, last_post_close_date=today) is False
+
+
+def test_post_close_trigger_skips_before_1600():
+    """16:00 전에는 trigger 안 함."""
+    now = _kst(2026, 5, 18, 15, 59)
+    assert _is_post_close_trigger(now, last_post_close_date=None) is False
+
+
+def test_post_close_trigger_fires_next_day_after_reset():
+    """다음 영업일이 되면 last_post_close_date < today.date() 이므로 다시 trigger."""
+    now = _kst(2026, 5, 19, 16, 0)
+    yesterday = _date(2026, 5, 18)
+    assert _is_post_close_trigger(now, last_post_close_date=yesterday) is True
+
+
+def test_post_close_trigger_skips_on_holiday():
+    """휴장일에는 trigger 안 함 (2026-05-05 어린이날)."""
+    now = _kst(2026, 5, 5, 16, 30)
+    assert _is_post_close_trigger(now, last_post_close_date=None) is False
+```
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+```
+python -m pytest ai_trade/tests/test_scheduler.py -v -k post_close
+```
+
+Expected: FAIL — `_is_post_close_trigger`가 신규 시그니처(`last_post_close_date` 인자) 미지원.
+
+- [ ] **Step 3: Modify scheduler.py:79-84**
+
+```python
+def _is_post_close_trigger(now: datetime, last_post_close_date) -> bool:
+    """16:00 KST 이후 오늘 아직 post-close cycle 안 돌렸으면 True (F3 상태기반).
+
+    Args:
+        now: 현재 KST datetime.
+        last_post_close_date: 마지막 post-close 실행 영업일 date 객체 (None=미실행).
+    """
+    if not _is_market_day(now):
+        return False
+    if now.time() < time(16, 0):
+        return False
+    today = now.date()
+    return last_post_close_date != today
+```
+
+- [ ] **Step 4: Run scheduler tests**
+
+```
+python -m pytest ai_trade/tests/test_scheduler.py -v
+```
+
+Expected: 신규 5개 PASS. 기존 test도 PASS (다른 함수 영향 없음).
+
+- [ ] **Step 5: Update pull_worker.py to track last_post_close_date**
+
+`ai_trade/pull_worker.py` 의 `poll_loop` (L18-58)을 다음으로 교체:
+
+```python
+async def poll_loop(
+    client: StockClient, state: PollState, shutdown: asyncio.Event,
+    kis_client: KISClient | None = None,
+    chronos=None,
+    dedup=None,
+    settings=None,
+) -> None:
+    """FastAPI lifespan 에서 asyncio.create_task 로 시작."""
+    logger.info("poll_loop started")
+    last_post_close_date = None
+    while not shutdown.is_set():
+        now = datetime.now(KST)
+        if _is_market_day(now) and _is_polling_window(now):
+            try:
+                await _run_polling_cycle(client, state, kis_client=kis_client)
+            except Exception:
+                logger.exception("poll cycle failed")
+            # Minute momentum 갱신 (매 cycle)
+            try:
+                update_minute_momentum_for_all(state)
+            except Exception:
+                logger.exception("minute momentum update failed")
+            # Post-close trigger (상태기반: 16:00 이후 + 오늘 미실행)
+            if (
+                _is_post_close_trigger(now, last_post_close_date)
+                and chronos is not None and kis_client is not None
+            ):
+                try:
+                    await _run_post_close_cycle(kis_client, chronos, state)
+                    last_post_close_date = now.date()
+                except Exception:
+                    logger.exception("post-close cycle failed")
+            # Phase 4: generate signals
+            if dedup is not None and settings is not None:
+                try:
+                    from ai_trade.signal_generator import generate_signals
+                    generate_signals(state, dedup, settings)
+                except Exception:
+                    logger.exception("generate_signals failed")
+        interval = _next_interval(now)
+        try:
+            await asyncio.wait_for(shutdown.wait(), timeout=interval)
+            break
+        except asyncio.TimeoutError:
+            continue
+    logger.info("poll_loop ended")
+```
+
+- [ ] **Step 6: Add pull_worker test**
+
+`ai_trade/tests/test_pull_worker.py` 파일 끝에 추가:
+
+```python
+from unittest.mock import AsyncMock, MagicMock
+from datetime import datetime as _dt
+from zoneinfo import ZoneInfo as _ZI
+import asyncio as _asyncio
+
+
+async def test_post_close_fires_at_1601_when_not_yet_today(monkeypatch):
+    """16:01에 깬 cycle도 post_close 안 돌렸으면 호출됨 (F3 회귀)."""
+    from ai_trade import pull_worker
+
+    _kst = _ZI("Asia/Seoul")
+    now_at_1601 = _dt(2026, 5, 18, 16, 1, tzinfo=_kst)
+
+    real_dt = _dt
+
+    class FrozenDateTime:
+        @staticmethod
+        def now(tz=None):
+            return now_at_1601
+
+    monkeypatch.setattr(pull_worker, "datetime", FrozenDateTime)
+    monkeypatch.setattr(
+        pull_worker, "_is_market_day", lambda n: True,
+    )
+    monkeypatch.setattr(
+        pull_worker, "_is_polling_window", lambda n: True,
+    )
+    monkeypatch.setattr(
+        pull_worker, "_next_interval", lambda n: 0.01,
+    )
+    monkeypatch.setattr(
+        pull_worker, "_run_polling_cycle", AsyncMock(),
+    )
+    monkeypatch.setattr(
+        pull_worker, "update_minute_momentum_for_all", lambda s: None,
+    )
+    post_close = AsyncMock()
+    monkeypatch.setattr(pull_worker, "_run_post_close_cycle", post_close)
+
+    state = MagicMock()
+    chronos = MagicMock()
+    kis = MagicMock()
+    shutdown = _asyncio.Event()
+
+    async def _stop_soon():
+        await _asyncio.sleep(0.05)
+        shutdown.set()
+
+    _asyncio.create_task(_stop_soon())
+    await pull_worker.poll_loop(
+        client=MagicMock(),
+        state=state,
+        shutdown=shutdown,
+        kis_client=kis,
+        chronos=chronos,
+        dedup=None,
+        settings=None,
+    )
+
+    assert post_close.await_count >= 1, "post-close가 16:01에 호출되지 않음 (F3 회귀)"
+```
+
+- [ ] **Step 7: Run pull_worker test**
+
+```
+python -m pytest ai_trade/tests/test_pull_worker.py::test_post_close_fires_at_1601_when_not_yet_today -v
+```
+
+Expected: PASS.
+
+- [ ] **Step 8: Run full ai_trade suite**
+
+```
+python -m pytest ai_trade/tests -q
+```
+
+Expected: 모두 PASS.
+
+- [ ] **Step 9: Commit**
+
+```bash
+git add ai_trade/scheduler.py ai_trade/pull_worker.py ai_trade/tests/test_scheduler.py ai_trade/tests/test_pull_worker.py
+git commit -m "fix(ai_trade): post-close trigger를 상태기반으로 변경 (F3)"
+```
+
+---
+
+## Task 4: F4 — Chronos confidence를 absolute spread 기반으로 통일
+
+**Files:**
+- Modify: `ai_trade/chronos_predictor.py:106, 127`
+- Test: `ai_trade/tests/test_chronos_predictor.py` (add 2 tests)
+
+**Why absolute:** Phase 4 spec amendment (web-ui commit 534ded5)가 absolute spread로 hard gate를 결정. confidence도 같은 철학으로. 새 산식: `conf = max(0, min(1, 1 - spread / SPREAD_THRESHOLD))` — spread가 0.6에 도달하면 conf=0, 0이면 conf=1.
+
+- [ ] **Step 1: Write the failing tests**
+
+기존 `ai_trade/tests/test_chronos_predictor.py` 끝에 추가 (파일이 없거나 비어있으면 신규):
+
+```python
+import numpy as np
+import pytest
+import torch
+
+
+@pytest.fixture
+def fake_pipeline():
+    """predict_quantiles만 stub하는 가짜 pipeline."""
+    class FakePipeline:
+        def __init__(self, q10_price, q50_price, q90_price):
+            self._q10, self._q50, self._q90 = q10_price, q50_price, q90_price
+        def predict_quantiles(self, contexts, prediction_length, quantile_levels):
+            n = len(contexts)
+            tensor = torch.tensor(
+                [[[self._q10, self._q50, self._q90]]] * n,
+                dtype=torch.float32,
+            )
+            return tensor, None
+    return FakePipeline
+
+
+def _make_predictor_with(pipeline_obj):
+    """ChronosPredictor 인스턴스 (실제 모델 안 부르고 pipeline만 주입)."""
+    from ai_trade.chronos_predictor import ChronosPredictor
+    p = ChronosPredictor.__new__(ChronosPredictor)
+    p._pipeline = pipeline_obj
+    p._device = "cpu"
+    return p
+
+
+def test_confidence_high_when_spread_near_zero(fake_pipeline):
+    """median≈0, spread≈0 (q10=q90=last_close)일 때 conf≈1 (F4)."""
+    last_close = 100000.0
+    p = _make_predictor_with(fake_pipeline(last_close, last_close, last_close))
+    ohlcv = {"A": [{"close": last_close}] * 30}
+    out = p.predict_batch(ohlcv)
+    assert out["A"].conf > 0.95, (
+        f"median≈0 + spread≈0인데 conf={out['A'].conf} (F4 회귀: relative spread로 폭증)"
+    )
+
+
+def test_confidence_drops_with_spread(fake_pipeline):
+    """spread 0.3일 때 conf≈0.5 (1 - 0.3/0.6 = 0.5)."""
+    last_close = 100000.0
+    # q10=85000 → -0.15, q90=115000 → 0.15, spread=0.30
+    p = _make_predictor_with(fake_pipeline(85000.0, 100000.0, 115000.0))
+    ohlcv = {"A": [{"close": last_close}] * 30}
+    out = p.predict_batch(ohlcv)
+    # 1 - 0.30/0.60 = 0.50
+    assert 0.45 < out["A"].conf < 0.55, (
+        f"absolute spread 0.30에서 conf={out['A'].conf} (expected ≈0.5)"
+    )
+
+
+def test_confidence_zero_at_threshold_spread(fake_pipeline):
+    """spread가 threshold(0.6) 이상이면 conf=0."""
+    last_close = 100000.0
+    # q10=70000 → -0.30, q90=130000 → 0.30, spread=0.60
+    p = _make_predictor_with(fake_pipeline(70000.0, 100000.0, 130000.0))
+    ohlcv = {"A": [{"close": last_close}] * 30}
+    out = p.predict_batch(ohlcv)
+    assert out["A"].conf < 0.05, (
+        f"spread=threshold에서 conf={out['A'].conf} (expected ≈0)"
+    )
+```
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+```
+python -m pytest ai_trade/tests/test_chronos_predictor.py -v -k confidence
+```
+
+Expected: `test_confidence_high_when_spread_near_zero` 가 FAIL — 현행 relative spread 산식 때문에 median≈0에서 conf가 0으로 폭락.
+
+- [ ] **Step 3: Fix chronos_predictor.py**
+
+`ai_trade/chronos_predictor.py` 상단에 상수 추가 (L13 근처):
+
+```python
+_SPREAD_THRESHOLD = 0.6  # F4: signal_generator hard gate와 동일 (absolute return spread)
+```
+
+L106 (modern API 경로) 변경:
+
+```python
+            # shape: [num_series, prediction_length, 3]
+            for i, ticker in enumerate(tickers):
+                q10_price, q50_price, q90_price = quantiles_np[i, 0, :]
+                last_close = daily_ohlcv_dict[ticker][-1]["close"]
+                median = float((q50_price - last_close) / last_close)
+                q10 = float((q10_price - last_close) / last_close)
+                q90 = float((q90_price - last_close) / last_close)
+                # F4: absolute spread (q90-q10) 기반 — signal_generator hard gate와 통일.
+                # median≈0 zero-shot 케이스에서 conf가 0으로 폭락하던 relative 산식 제거.
+                spread = q90 - q10
+                conf = float(max(0.0, min(1.0, 1.0 - spread / _SPREAD_THRESHOLD)))
+                results[ticker] = ChronosPrediction(
+                    median=median, q10=q10, q90=q90, conf=conf, as_of=now_iso,
+                )
+            return results
+```
+
+L127 (legacy API 경로) 동일하게 변경:
+
+```python
+            spread = q90 - q10
+            conf = float(max(0.0, min(1.0, 1.0 - spread / _SPREAD_THRESHOLD)))
+```
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+```
+python -m pytest ai_trade/tests/test_chronos_predictor.py -v
+```
+
+Expected: 신규 3개 모두 PASS. 기존 test도 PASS.
+
+- [ ] **Step 5: Run full ai_trade suite**
+
+```
+python -m pytest ai_trade/tests -q
+```
+
+Expected: 모두 PASS. signal_generator 테스트(`_compute_buy_confidence` 가 `pred["conf"]` 사용) 도 영향 받을 수 있으니 주시.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add ai_trade/chronos_predictor.py ai_trade/tests/test_chronos_predictor.py
+git commit -m "fix(ai_trade): Chronos confidence를 absolute spread 기반으로 통일 (F4)"
+```
+
+---
+
+## Task 5: 전체 회귀 확인 + push
+
+- [ ] **Step 1: Run full ai_trade suite + count**
+
+```
+python -m pytest ai_trade/tests -v
+```
+
+Expected:
+- 기존 56 tests + 신규 (config 2 + kis_client 1 + scheduler 5 + pull_worker 1 + chronos_predictor 3) = **68 tests** 정도 PASS.
+
+- [ ] **Step 2: Quick sanity — server boot smoke test (시간 허용 시)**
+
+```
+cd ai_trade && python -c "from main import app; print('app import OK')"
+```
+
+Expected: no import errors.
+
+- [ ] **Step 3: Push**
+
+```bash
+git push origin main
+```
+
+---
+
+## Self-Review Checklist
+
+이 plan을 다 작성한 뒤 다음을 확인:
+
+1. **F1**: config.py default + 2 test (default + env override) ✅
+2. **F2**: `_throttle_lock` 추가 + 1 concurrent test ✅
+3. **F3**: `_is_post_close_trigger(now, last_post_close_date)` 시그니처 변경 + `poll_loop` 상태 추적 + 5 scheduler test + 1 pull_worker test ✅
+4. **F4**: `_SPREAD_THRESHOLD=0.6` 상수 + 두 분기(modern + legacy) 모두 absolute spread 적용 + 3 chronos_predictor test ✅
+
+**누락 가능 항목**:
+- `test_main.py` 가 `v1_token_path` default를 직접 검증한다면 Task 1에서 같이 갱신. 위 patch는 Settings 객체 통해서만 다루므로 영향 없음(검증 완료).
+- Task 3 pull_worker test의 `FrozenDateTime.now`는 `datetime.now(KST)` 호출만 stub함. 다른 datetime 사용 부분 영향 없음 (verified L28).
+- Task 4 test는 ChronosPredictor `__new__`로 우회 — 실제 HuggingFace 모델 로딩 안 함.
+
+---
+
+## Execution Handoff
+
+**Plan complete and saved to `docs/superpowers/plans/2026-05-25-ai-trade-hotfix.md`.**
+
+두 가지 실행 옵션:
+
+**1. Subagent-Driven (recommended)** — task 별 fresh subagent dispatch + two-stage review. F2/F3 같이 미묘한 동시성/상태 변경에 유리.
+
+**2. Inline Execution** — 현 세션에서 직접 task별 진행 + checkpoint.
+
+박재오 결정 대기.

docs/superpowers/plans/2026-05-25-render-queue-reliability.md

@@ -0,0 +1,704 @@
+# Render Queue Reliability — Code Review F6 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:** 4개 render worker(insta/music/video/image-render)가 BLPOP 직후 crash 시 작업 손실되는 문제 해결. BLMOVE(또는 BRPOPLPUSH)로 atomic dequeue + processing list 패턴 + startup recovery + retry/dead-letter.
+
+**Architecture:**
+1. 각 worker가 unique `worker_id` 보유: `<queue>-<hostname>-<pid>` (env로 override 가능).
+2. atomic dequeue: `BLMOVE queue:<x>-render processing:<x>-render:<worker_id> RIGHT LEFT 5` — 5초 timeout. (`BRPOPLPUSH`는 Redis 6.2+ deprecated, `BLMOVE`가 후속).
+3. 작업 성공: `LREM processing:<x>-render:<worker_id> 1 <payload>` — 정확 1개 제거.
+4. 작업 실패: payload에 `attempts` counter 증가시켜 main queue 끝으로 LPUSH; 한계(기본 3) 초과 시 `dead_letter:<queue>` 로 이동.
+5. **Startup recovery**: worker 시작 시 자신의 processing list가 비어있지 않으면 → 모두 main queue로 되돌림 (재시도). attempts 증가.
+6. NAS측 producer는 무변경 (LPUSH 그대로). 단, payload schema에 `attempts: int` (optional) 필드 명시 — producer는 안 채워도 worker가 default 0으로.
+
+**Shared module 전략:** 4개 worker가 동일 패턴이므로 `services/_shared/reliable_queue.py` 1개 만들고 각 Dockerfile에서 `COPY services/_shared /app/_shared` 후 `from _shared.reliable_queue import ReliableQueue`. compose entry/dockerfile 변경 4건. (DRY > inline 4중복.)
+
+**Tech Stack:** Python 3.12, redis.asyncio 5.x, fakeredis (pytest dep), pytest-asyncio.
+
+**Working directory:** `C:\Users\jaeoh\Desktop\workspace\web-ai`.
+
+---
+
+## File Map
+
+| 파일 | 변경 | 책임 |
+|------|------|------|
+| `services/_shared/__init__.py` | Create | namespace package |
+| `services/_shared/reliable_queue.py` | Create | `ReliableQueue` 클래스 — dequeue, ack, fail, recover |
+| `services/_shared/tests/test_reliable_queue.py` | Create | fakeredis 단위 테스트 6개 |
+| `services/_shared/requirements.txt` | Create | redis>=5.0, fakeredis (test only) |
+| `services/insta-render/Dockerfile` | Modify | `COPY services/_shared /app/_shared` + PYTHONPATH |
+| `services/insta-render/worker.py` | Modify L1~ | BLPOP → ReliableQueue 사용 |
+| `services/insta-render/tests/test_worker.py` | Append | 1 integration test (recovery) |
+| `services/music-render/Dockerfile` | Modify | shared copy |
+| `services/music-render/worker.py` | Modify | ReliableQueue 사용 |
+| `services/music-render/tests/test_worker.py` | Append | recovery test |
+| `services/video-render/Dockerfile` | Modify | shared copy |
+| `services/video-render/worker.py` | Modify | ReliableQueue 사용 |
+| `services/video-render/tests/test_worker.py` | Append | recovery test |
+| `services/image-render/Dockerfile` | Modify | shared copy |
+| `services/image-render/worker.py` | Modify | ReliableQueue 사용 |
+| `services/image-render/tests/test_worker.py` | Append | recovery test |
+| `services/docker-compose.yml` (있다면) | Verify | build context가 services/ 루트 포함하는지 |
+
+---
+
+## Task 1: ReliableQueue 공유 모듈 작성
+
+**Files:**
+- Create: `services/_shared/__init__.py`
+- Create: `services/_shared/reliable_queue.py`
+- Create: `services/_shared/tests/__init__.py`
+- Create: `services/_shared/tests/test_reliable_queue.py`
+- Create: `services/_shared/requirements.txt`
+
+- [ ] **Step 1: Create namespace package**
+
+```python
+# services/_shared/__init__.py
+```
+(빈 파일)
+
+```python
+# services/_shared/tests/__init__.py
+```
+(빈 파일)
+
+- [ ] **Step 2: Write failing tests first**
+
+```python
+# services/_shared/tests/test_reliable_queue.py
+"""F6 — ReliableQueue: atomic dequeue + recovery + retry."""
+import json
+
+import fakeredis.aioredis
+import pytest
+
+from _shared.reliable_queue import ReliableQueue
+
+
+@pytest.fixture
+async def redis():
+    r = fakeredis.aioredis.FakeRedis(decode_responses=False)
+    yield r
+    await r.flushall()
+    await r.aclose()
+
+
+async def test_dequeue_atomically_moves_to_processing(redis):
+    """BLMOVE: queue → processing 원자적 이동."""
+    q = ReliableQueue(redis, queue_key="queue:test", worker_id="w1")
+    await redis.lpush("queue:test", json.dumps({"task_id": "t1"}).encode())
+    payload, raw = await q.dequeue(timeout=1)
+    assert payload["task_id"] == "t1"
+    # main queue는 비어있고, processing list에 들어있어야 함
+    assert await redis.llen("queue:test") == 0
+    assert await redis.llen("processing:queue:test:w1") == 1
+
+
+async def test_dequeue_returns_none_on_timeout(redis):
+    q = ReliableQueue(redis, queue_key="queue:test", worker_id="w1")
+    result = await q.dequeue(timeout=1)
+    assert result is None
+
+
+async def test_ack_removes_from_processing(redis):
+    q = ReliableQueue(redis, queue_key="queue:test", worker_id="w1")
+    await redis.lpush("queue:test", json.dumps({"task_id": "t1"}).encode())
+    payload, raw = await q.dequeue(timeout=1)
+    await q.ack(raw)
+    assert await redis.llen("processing:queue:test:w1") == 0
+
+
+async def test_recover_returns_orphaned_to_main_queue(redis):
+    """startup recovery: 잔존 processing list 항목을 main queue로 되돌림."""
+    # 이전 crash 시뮬레이션: processing list에 잔존
+    orphan = json.dumps({"task_id": "t1", "attempts": 0}).encode()
+    await redis.lpush("processing:queue:test:w1", orphan)
+    q = ReliableQueue(redis, queue_key="queue:test", worker_id="w1")
+    recovered = await q.recover()
+    assert recovered == 1
+    assert await redis.llen("processing:queue:test:w1") == 0
+    # 다시 dequeue 가능
+    payload, raw = await q.dequeue(timeout=1)
+    assert payload["task_id"] == "t1"
+    assert payload["attempts"] == 1  # incremented on recover
+
+
+async def test_fail_below_max_attempts_returns_to_main_queue(redis):
+    q = ReliableQueue(redis, queue_key="queue:test", worker_id="w1", max_attempts=3)
+    await redis.lpush("queue:test", json.dumps({"task_id": "t1", "attempts": 0}).encode())
+    payload, raw = await q.dequeue(timeout=1)
+    await q.fail(raw, payload)
+    assert await redis.llen("processing:queue:test:w1") == 0
+    assert await redis.llen("queue:test") == 1
+    # attempts 증가됐는지
+    requeued_raw = await redis.lindex("queue:test", 0)
+    requeued = json.loads(requeued_raw)
+    assert requeued["attempts"] == 1
+
+
+async def test_fail_at_max_attempts_moves_to_dead_letter(redis):
+    q = ReliableQueue(redis, queue_key="queue:test", worker_id="w1", max_attempts=3)
+    await redis.lpush(
+        "queue:test", json.dumps({"task_id": "t1", "attempts": 2}).encode()
+    )
+    payload, raw = await q.dequeue(timeout=1)
+    await q.fail(raw, payload)
+    # attempts 2 → 3 (== max) → dead-letter
+    assert await redis.llen("queue:test") == 0
+    assert await redis.llen("processing:queue:test:w1") == 0
+    assert await redis.llen("dead_letter:queue:test") == 1
+```
+
+- [ ] **Step 3: Add requirements**
+
+```text
+# services/_shared/requirements.txt
+redis>=5.0.0
+```
+
+별도 dev requirements (test):
+
+```text
+# services/_shared/tests/requirements-dev.txt (optional)
+fakeredis>=2.20.0
+pytest>=8.0.0
+pytest-asyncio>=0.23.0
+```
+
+- [ ] **Step 4: Run tests to verify they fail**
+
+```
+cd services/_shared && python -m pytest tests/ -v
+```
+
+Expected: ImportError — reliable_queue.py 미존재.
+
+- [ ] **Step 5: Write reliable_queue.py**
+
+```python
+# services/_shared/reliable_queue.py
+"""F6 — Reliable Redis queue with processing list + recovery + retry.
+
+Pattern: BLMOVE main → processing (atomic), then either ack (LREM processing) or
+fail (LREM processing + re-enqueue or dead-letter).
+
+Startup recovery: any items left in the worker's processing list from a previous
+crash are pushed back to main queue (with attempts incremented).
+"""
+from __future__ import annotations
+
+import json
+import logging
+import os
+import socket
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+def default_worker_id(queue_key: str) -> str:
+    """env > hostname-pid."""
+    explicit = os.getenv("WORKER_ID")
+    if explicit:
+        return explicit
+    return f"{queue_key}-{socket.gethostname()}-{os.getpid()}"
+
+
+class ReliableQueue:
+    """Wraps a redis client to provide BLMOVE-backed atomic dequeue +
+    processing list + retry/dead-letter.
+
+    Producer side stays unchanged: LPUSH queue:<x> <json payload>.
+    Worker side: dequeue() → process → ack(raw) on success or fail(raw, payload) on error.
+    Startup: await queue.recover() to re-enqueue orphans.
+    """
+
+    def __init__(
+        self,
+        redis,
+        queue_key: str,
+        worker_id: Optional[str] = None,
+        max_attempts: int = 3,
+    ):
+        self._redis = redis
+        self._queue_key = queue_key
+        self._worker_id = worker_id or default_worker_id(queue_key)
+        self._processing_key = f"processing:{queue_key}:{self._worker_id}"
+        self._dead_letter_key = f"dead_letter:{queue_key}"
+        self._max_attempts = max_attempts
+
+    @property
+    def processing_key(self) -> str:
+        return self._processing_key
+
+    async def dequeue(self, timeout: int = 5) -> Optional[tuple[dict, bytes]]:
+        """Atomically move 1 item from main queue tail to processing head.
+
+        Returns (parsed_dict, raw_bytes) or None on timeout.
+        Caller MUST call ack(raw) on success or fail(raw, payload) on error.
+        """
+        raw = await self._redis.blmove(
+            self._queue_key, self._processing_key,
+            timeout=timeout, src="RIGHT", dest="LEFT",
+        )
+        if raw is None:
+            return None
+        try:
+            payload = json.loads(raw)
+        except json.JSONDecodeError:
+            logger.error("invalid payload on dequeue, moving to dead-letter: %r", raw[:200])
+            await self._redis.lrem(self._processing_key, 1, raw)
+            await self._redis.lpush(self._dead_letter_key, raw)
+            return None
+        return payload, raw
+
+    async def ack(self, raw: bytes) -> None:
+        """Successful processing — remove from processing list."""
+        removed = await self._redis.lrem(self._processing_key, 1, raw)
+        if removed == 0:
+            logger.warning("ack on missing payload (already removed?): %r", raw[:100])
+
+    async def fail(self, raw: bytes, payload: dict) -> None:
+        """Failed processing — remove from processing list and either re-enqueue or dead-letter."""
+        await self._redis.lrem(self._processing_key, 1, raw)
+        attempts = int(payload.get("attempts", 0)) + 1
+        if attempts >= self._max_attempts:
+            payload["attempts"] = attempts
+            await self._redis.lpush(self._dead_letter_key, json.dumps(payload).encode())
+            logger.error(
+                "task moved to dead-letter after %d attempts: task_id=%s",
+                attempts, payload.get("task_id"),
+            )
+            return
+        payload["attempts"] = attempts
+        await self._redis.lpush(self._queue_key, json.dumps(payload).encode())
+        logger.info(
+            "task re-enqueued (attempt %d/%d): task_id=%s",
+            attempts, self._max_attempts, payload.get("task_id"),
+        )
+
+    async def recover(self) -> int:
+        """Startup: move all orphans from this worker's processing list back to main queue.
+
+        Increments attempts counter (orphan == implicit failure).
+        Returns count of recovered items.
+        """
+        count = 0
+        while True:
+            raw = await self._redis.lpop(self._processing_key)
+            if raw is None:
+                break
+            try:
+                payload = json.loads(raw)
+            except json.JSONDecodeError:
+                await self._redis.lpush(self._dead_letter_key, raw)
+                count += 1
+                continue
+            payload["attempts"] = int(payload.get("attempts", 0)) + 1
+            if payload["attempts"] >= self._max_attempts:
+                await self._redis.lpush(self._dead_letter_key, json.dumps(payload).encode())
+            else:
+                await self._redis.lpush(self._queue_key, json.dumps(payload).encode())
+            count += 1
+        if count:
+            logger.info("recovered %d orphaned items for worker %s", count, self._worker_id)
+        return count
+```
+
+**참고: redis-py blmove API**: `client.blmove(first_list, second_list, timeout, src=..., dest=...)`. timeout=0 은 block forever. payload는 bytes로 받음 (`decode_responses=False` 가정).
+
+- [ ] **Step 6: Run tests to verify they pass**
+
+```
+cd services/_shared && python -m pytest tests/ -v
+```
+
+Expected: 6 PASS.
+
+만약 ImportError (`fakeredis` 미설치) 발생 시:
+
+```
+python -m pip install fakeredis pytest-asyncio
+```
+
+또한 `pytest.ini` 또는 `conftest.py`에 `asyncio_mode = "auto"` 필요. 신규 conftest:
+
+```python
+# services/_shared/tests/conftest.py
+import pytest
+pytest_plugins = ["pytest_asyncio"]
+
+
+def pytest_collection_modifyitems(config, items):
+    for item in items:
+        if "asyncio" in item.fixturenames or item.get_closest_marker("asyncio") is not None:
+            continue
+        # auto-mark all async tests
+        if item.function.__name__.startswith("test_"):
+            import asyncio, inspect
+            if inspect.iscoroutinefunction(item.function):
+                item.add_marker(pytest.mark.asyncio)
+```
+
+또는 더 간단히 `services/_shared/pytest.ini`:
+
+```ini
+[pytest]
+asyncio_mode = auto
+```
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add services/_shared/
+git commit -m "feat(services): _shared/reliable_queue 신설 — BLMOVE + processing list + retry (F6 part 1)"
+```
+
+---
+
+## Task 2: insta-render에 ReliableQueue 적용
+
+**Files:**
+- Modify: `services/insta-render/Dockerfile`
+- Modify: `services/insta-render/worker.py`
+- Modify: `services/insta-render/tests/test_worker.py` (append)
+
+- [ ] **Step 1: Update Dockerfile**
+
+`services/insta-render/Dockerfile` 에 `_shared` 복사 추가. 기존 Dockerfile 패턴을 먼저 읽고, `COPY services/insta-render /app` 같은 라인이 있다면 그 위 또는 옆에:
+
+```dockerfile
+COPY services/_shared /app/_shared
+ENV PYTHONPATH=/app:/app/_shared:${PYTHONPATH}
+```
+
+build context가 `services/` 루트여야 함. compose에서 `build: { context: ./services, dockerfile: insta-render/Dockerfile }` 인지 확인 — 아니라면 context 조정 필요.
+
+- [ ] **Step 2: Modify worker.py — failing test first**
+
+`services/insta-render/tests/test_worker.py` 끝에 추가:
+
+```python
+import json
+from unittest.mock import AsyncMock, patch
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_worker_calls_ack_on_success():
+    """성공 시 ack() 호출 (F6)."""
+    import worker
+    fake_payload = {"task_id": "t1", "job_type": "card_generation", "params": {}}
+    fake_raw = json.dumps(fake_payload).encode()
+
+    fake_queue = AsyncMock()
+    fake_queue.dequeue = AsyncMock(side_effect=[(fake_payload, fake_raw), None])
+    fake_queue.ack = AsyncMock()
+    fake_queue.fail = AsyncMock()
+    fake_queue.recover = AsyncMock(return_value=0)
+
+    with patch.object(worker, "ReliableQueue", return_value=fake_queue), \
+         patch.object(worker, "_dispatch") as disp:
+        # poll_once로 1 cycle만 실행 (실제 loop 끊기 위해)
+        await worker.poll_once(fake_queue)
+        disp.assert_called_once()
+        fake_queue.ack.assert_called_once_with(fake_raw)
+        fake_queue.fail.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_worker_calls_fail_on_dispatch_exception():
+    """dispatch 예외 시 fail() 호출 — 작업 손실 안 됨 (F6)."""
+    import worker
+    fake_payload = {"task_id": "t2", "job_type": "card_generation", "params": {}}
+    fake_raw = json.dumps(fake_payload).encode()
+
+    fake_queue = AsyncMock()
+    fake_queue.dequeue = AsyncMock(return_value=(fake_payload, fake_raw))
+    fake_queue.ack = AsyncMock()
+    fake_queue.fail = AsyncMock()
+
+    with patch.object(worker, "_dispatch", side_effect=RuntimeError("boom")):
+        await worker.poll_once(fake_queue)
+        fake_queue.fail.assert_called_once_with(fake_raw, fake_payload)
+        fake_queue.ack.assert_not_called()
+```
+
+- [ ] **Step 3: Run test to fail**
+
+```
+cd services/insta-render && python -m pytest tests/ -v -k "ack_on_success or fail_on_dispatch"
+```
+
+Expected: AttributeError (`worker.poll_once` 미존재, `worker.ReliableQueue` 미존재).
+
+- [ ] **Step 4: Rewrite insta-render worker.py**
+
+```python
+"""Redis 기반 worker — F6 신뢰성 패턴 적용 (BLMOVE + processing list + retry)."""
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import sys
+
+import redis.asyncio as aioredis
+
+from _shared.reliable_queue import ReliableQueue
+from nas_client import webhook_update_task
+# 기존 dispatch 대상 import 유지
+from card_renderer import render_card
+
+logger = logging.getLogger(__name__)
+
+REDIS_URL = os.getenv("REDIS_URL", "redis://192.168.45.54:6379")
+QUEUE_KEY = "queue:insta-render"
+PAUSED_KEY = "queue:paused"
+
+
+_DISPATCH_TABLE = {
+    "card_generation": "render_card",
+}
+
+
+def _dispatch(payload: dict) -> None:
+    job_type = payload.get("job_type", "")
+    task_id = payload.get("task_id", "")
+    params = payload.get("params", {})
+    fn_name = _DISPATCH_TABLE.get(job_type)
+    if fn_name is None:
+        logger.error("unknown job_type=%s task=%s", job_type, task_id)
+        webhook_update_task(task_id, "failed", 0, "", error=f"unknown job_type: {job_type}")
+        return
+    try:
+        fn = getattr(sys.modules[__name__], fn_name)
+    except AttributeError:
+        webhook_update_task(task_id, "failed", 0, "", error=f"internal dispatch error: {fn_name}")
+        return
+    fn(task_id, params)
+
+
+async def poll_once(queue: ReliableQueue) -> bool:
+    """1 cycle: dequeue → dispatch → ack/fail. Returns True if a job was handled."""
+    result = await queue.dequeue(timeout=5)
+    if result is None:
+        return False
+    payload, raw = result
+    try:
+        await asyncio.to_thread(_dispatch, payload)
+    except Exception:
+        logger.exception("dispatch failed task_id=%s", payload.get("task_id"))
+        await queue.fail(raw, payload)
+        return True
+    await queue.ack(raw)
+    return True
+
+
+async def worker_loop():
+    redis = aioredis.from_url(REDIS_URL, decode_responses=False)
+    queue = ReliableQueue(redis, queue_key=QUEUE_KEY)
+    logger.info("insta-render worker started worker_id=%s", queue._worker_id)
+    # F6: startup recovery
+    try:
+        recovered = await queue.recover()
+        if recovered:
+            logger.info("recovered %d orphaned items at startup", recovered)
+    except Exception:
+        logger.exception("startup recover failed")
+    while True:
+        try:
+            paused = await redis.get(PAUSED_KEY)
+            if paused == b"1":
+                await asyncio.sleep(10)
+                continue
+            await poll_once(queue)
+        except asyncio.CancelledError:
+            logger.info("worker_loop cancelled")
+            raise
+        except Exception:
+            logger.exception("worker_loop iteration 실패, 5초 후 재시도")
+            await asyncio.sleep(5)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+    asyncio.run(worker_loop())
+```
+
+**NOTE**: 기존 `insta-render/worker.py`의 dispatch table·import는 실제 파일을 보고 매핑 유지. 위 예시는 minimal — job_type / function 이름은 기존 파일과 맞춰야 함. 변경 전 `Read services/insta-render/worker.py`로 정확한 dispatch table 확인할 것.
+
+- [ ] **Step 5: Run tests**
+
+```
+cd services/insta-render && python -m pytest tests/ -v
+```
+
+Expected: 신규 2 PASS, 기존 PASS (dispatch table test 등).
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add services/insta-render/
+git commit -m "fix(insta-render): F6 ReliableQueue 적용 — BLMOVE + ack/fail (F6 part 2)"
+```
+
+---
+
+## Task 3: music-render에 동일 적용
+
+**Files:**
+- Modify: `services/music-render/Dockerfile`, `worker.py`
+- Modify: `services/music-render/tests/test_worker.py` (append)
+
+- [ ] **Step 1: Dockerfile에 `COPY services/_shared` 추가**
+- [ ] **Step 2: Test 추가 (Task 2 패턴 동일, 단 dispatch target은 `run_suno_generation` 등 기존 패턴)**
+
+```python
+@pytest.mark.asyncio
+async def test_music_worker_ack_on_success():
+    import worker
+    payload = {"task_id": "t1", "job_type": "suno_generation", "params": {}}
+    raw = json.dumps(payload).encode()
+    fake_queue = AsyncMock()
+    fake_queue.dequeue = AsyncMock(return_value=(payload, raw))
+    fake_queue.ack = AsyncMock()
+    with patch.object(worker, "_dispatch"):
+        await worker.poll_once(fake_queue)
+        fake_queue.ack.assert_called_once_with(raw)
+
+
+@pytest.mark.asyncio
+async def test_music_worker_fail_on_exception():
+    import worker
+    payload = {"task_id": "t2", "job_type": "suno_generation", "params": {}}
+    raw = json.dumps(payload).encode()
+    fake_queue = AsyncMock()
+    fake_queue.dequeue = AsyncMock(return_value=(payload, raw))
+    fake_queue.fail = AsyncMock()
+    with patch.object(worker, "_dispatch", side_effect=RuntimeError("x")):
+        await worker.poll_once(fake_queue)
+        fake_queue.fail.assert_called_once_with(raw, payload)
+```
+
+- [ ] **Step 3: Run test to fail**
+- [ ] **Step 4: Rewrite music-render worker.py — `worker_loop` 구조는 insta-render와 동일, `_dispatch` + `_DISPATCH_TABLE`은 기존 12개 함수 그대로 유지**
+- [ ] **Step 5: Run tests**
+- [ ] **Step 6: Commit**
+
+```bash
+git add services/music-render/
+git commit -m "fix(music-render): F6 ReliableQueue 적용 (F6 part 3)"
+```
+
+---
+
+## Task 4: video-render에 동일 적용
+
+(Task 3와 동일 패턴 — sora/veo/kling/seedance 4 provider table 유지)
+
+- [ ] **Step 1: Dockerfile 수정**
+- [ ] **Step 2: 신규 test 2개 추가 (`test_video_worker_ack_on_success`, `test_video_worker_fail_on_exception`) — job_type은 `sora_generation`**
+- [ ] **Step 3: Run failing test**
+- [ ] **Step 4: Rewrite worker.py — 동일 패턴**
+- [ ] **Step 5: Run tests**
+- [ ] **Step 6: Commit**
+
+```bash
+git add services/video-render/
+git commit -m "fix(video-render): F6 ReliableQueue 적용 (F6 part 4)"
+```
+
+---
+
+## Task 5: image-render에 동일 적용
+
+(gpt_image / nano_banana / flux 3 provider table 유지)
+
+- [ ] **Step 1-6: Task 3/4 동일 패턴**
+
+```bash
+git add services/image-render/
+git commit -m "fix(image-render): F6 ReliableQueue 적용 (F6 part 5)"
+```
+
+---
+
+## Task 6: 운영 검증 + push
+
+- [ ] **Step 1: 전체 services test 실행**
+
+```
+cd services && for d in _shared insta-render music-render video-render image-render; do
+  echo "--- $d ---"
+  (cd $d && python -m pytest tests/ -q) || true
+done
+```
+
+(또는 PowerShell:)
+
+```powershell
+foreach ($d in @("_shared","insta-render","music-render","video-render","image-render")) {
+  Write-Output "--- $d ---"
+  Push-Location services/$d
+  python -m pytest tests/ -q
+  Pop-Location
+}
+```
+
+Expected: 4개 worker 각 신규 2개 + _shared 6개 + 기존 test 전부 PASS.
+
+- [ ] **Step 2: Docker build 시뮬 (옵션, 시간 허용 시)**
+
+```
+cd services && docker compose build insta-render music-render video-render image-render
+```
+
+Expected: build context에 `_shared` 포함됨 검증.
+
+- [ ] **Step 3: Push**
+
+```bash
+git push origin main
+```
+
+- [ ] **Step 4: 운영 deploy 시 주의사항 (수동)**
+
+NAS에서 컨테이너 재배포 시:
+1. `redis-cli -h 192.168.45.54 KEYS 'processing:*'` 로 기존 orphan 확인 — 있다면 worker_id 다르면 안 잡힘. 수동으로 `LMOVE` 해야 할 수도 있음.
+2. `redis-cli -h 192.168.45.54 KEYS 'dead_letter:*'` 로 dead-letter 모니터 — 누적되면 alerting 필요.
+3. WORKER_ID env로 unique 하게 (`WORKER_ID=insta-render-prod-1` 등) 권장 — hostname이 컨테이너 재기동 시 바뀌면 orphan 추적 안 됨.
+
+---
+
+## Self-Review
+
+1. **atomic dequeue**: `BLMOVE` 단일 명령 — Redis 단일 트랜잭션 ✅
+2. **ack on success**: `LREM processing 1 raw` — 정확 1개 ✅
+3. **fail with retry**: attempts < max → 재큐, attempts >= max → dead-letter ✅
+4. **startup recovery**: orphan 자동 재큐 (attempts 증가) ✅
+5. **4 worker 적용**: insta/music/video/image 동일 패턴 ✅
+6. **NAS producer 호환**: LPUSH 그대로, payload schema에 attempts 선택적 ✅
+
+**미커버 (의도적)**:
+- dead-letter monitor/alert — 운영 작업 (CHECK_POINT 백로그)
+- worker_id env 미설정 시 hostname 변경 시 orphan 분실 — 운영 가이드에 명시
+
+**가정 검증**:
+- `redis-py.aioredis.blmove` 시그니처: `(first_list, second_list, timeout, src='LEFT', dest='RIGHT')`. redis>=5.0 권장.
+- fakeredis: `fakeredis.aioredis.FakeRedis` (>=2.20.0) 가 BLMOVE 지원함 — 미지원 시 plan 적용 전 검증.
+
+---
+
+## Execution Handoff
+
+**Plan complete and saved to `docs/superpowers/plans/2026-05-25-render-queue-reliability.md`.**
+
+**1. Subagent-Driven (recommended)** — Task 별 fresh subagent. 4개 worker는 패턴 같으나 dispatch table은 각 worker 고유 — subagent가 정확히 일관성 유지하도록 review checkpoint.
+
+**2. Inline Execution** — 현 세션 실행.
+
+박재오 결정 대기. Plan 1·2 마친 후 진입 권장 (작업량 가장 큼 — 4개 worker × 약 1시간 = 4시간).

docs/superpowers/plans/2026-05-25-state-signals-lifecycle.md

@@ -0,0 +1,593 @@
+# state.signals Lifecycle — Code Review F5 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:** `state.signals` 가 무한 dict 누적되는 문제를 해결. expires_at + cycle_id 부착해서 Phase 5 consumer (agent-office `/signal`) 가 stale 신호를 안전하게 무시할 수 있게.
+
+**Architecture:**
+1. `Signal` dict에 `expires_at: ISO str`, `cycle_id: int` 필드 추가.
+2. `PollState.signal_cycle_id: int` (process 단위 auto-increment).
+3. `generate_signals(state, dedup, settings)` 진입마다 `cycle_id += 1`.
+4. emit하는 모든 signal에 `expires_at = as_of + SIGNAL_TTL_SECONDS`, `cycle_id = state.signal_cycle_id` 부착.
+5. `state.purge_expired_signals(now)` helper — 매 cycle 끝에 호출하여 만료된 항목 제거.
+6. `state.get_active_signals(now) → list[dict]` — Phase 5 consumer가 호출할 read API. 만료된 것 제외.
+
+**Tech Stack:** Python 3.12, asyncio, pytest. 기존 cycle 흐름과 호환되도록 generate_signals 인터페이스는 그대로.
+
+**Why expires_at + cycle_id (not pop-on-read):** consumer가 polling 실패해도 신호 손실 없음. cycle_id로 "이번 cycle에 새로 emit된 신호" 식별 가능 → Phase 5에서 incremental fetch 가능.
+
+**Working directory:** `C:\Users\jaeoh\Desktop\workspace\web-ai`.
+
+**Test runner:** `python -m pytest ai_trade/tests -q` (또는 `py -3.12 -m`). 환경 부재 시 plan 진행 중단.
+
+---
+
+## File Map
+
+| 파일 | 변경 | 책임 |
+|------|------|------|
+| `ai_trade/config.py` | Add 1 field | `signal_ttl_seconds: int` (default 300) |
+| `ai_trade/state.py` | Modify | `signal_cycle_id: int`, helper 2개 (`get_active_signals`, `purge_expired_signals`) |
+| `ai_trade/signal_generator.py` | Modify L22-50, 133, 99-111, 174-186 | cycle_id 증가 + expires_at/cycle_id 부착 |
+| `ai_trade/pull_worker.py` | Modify L46-51 근처 | cycle 끝에 purge 호출 |
+| `ai_trade/tests/test_state_signals_lifecycle.py` | Create | 5 test (expires, cycle_id, purge, active list) |
+| `ai_trade/tests/test_signal_generator.py` | Modify | 기존 emit test에 expires_at/cycle_id 필드 검증 추가 |
+
+---
+
+## Task 1: PollState에 cycle_id + lifecycle helper 추가
+
+**Files:**
+- Modify: `ai_trade/state.py`
+- Test: `ai_trade/tests/test_state_signals_lifecycle.py` (Create)
+
+- [ ] **Step 1: Write the failing test**
+
+```python
+# ai_trade/tests/test_state_signals_lifecycle.py
+"""F5 — state.signals lifecycle (expires_at + cycle_id)."""
+from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo
+
+import pytest
+
+from ai_trade.state import PollState
+
+KST = ZoneInfo("Asia/Seoul")
+
+
+def test_initial_signal_cycle_id_is_zero():
+    state = PollState()
+    assert state.signal_cycle_id == 0
+
+
+def test_get_active_signals_excludes_expired():
+    state = PollState()
+    now = datetime(2026, 5, 25, 10, 0, tzinfo=KST)
+    future = (now + timedelta(seconds=300)).isoformat()
+    past = (now - timedelta(seconds=60)).isoformat()
+    state.signals = {
+        "A": {"ticker": "A", "expires_at": future, "cycle_id": 1, "action": "buy"},
+        "B": {"ticker": "B", "expires_at": past,   "cycle_id": 1, "action": "buy"},
+    }
+    active = state.get_active_signals(now)
+    tickers = [s["ticker"] for s in active]
+    assert "A" in tickers
+    assert "B" not in tickers
+
+
+def test_get_active_signals_treats_missing_expires_as_expired():
+    """expires_at 없는 legacy 신호는 expired로 간주."""
+    state = PollState()
+    now = datetime(2026, 5, 25, 10, 0, tzinfo=KST)
+    state.signals = {"C": {"ticker": "C", "action": "buy"}}
+    assert state.get_active_signals(now) == []
+
+
+def test_purge_expired_signals_removes_expired():
+    state = PollState()
+    now = datetime(2026, 5, 25, 10, 0, tzinfo=KST)
+    future = (now + timedelta(seconds=300)).isoformat()
+    past = (now - timedelta(seconds=60)).isoformat()
+    state.signals = {
+        "A": {"ticker": "A", "expires_at": future, "cycle_id": 1},
+        "B": {"ticker": "B", "expires_at": past, "cycle_id": 1},
+    }
+    state.purge_expired_signals(now)
+    assert "A" in state.signals
+    assert "B" not in state.signals
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+```
+python -m pytest ai_trade/tests/test_state_signals_lifecycle.py -v
+```
+
+Expected: `AttributeError: signal_cycle_id` 또는 `get_active_signals` 미구현.
+
+- [ ] **Step 3: Modify state.py**
+
+```python
+"""PollState — process-wide singleton."""
+from collections import deque
+from dataclasses import dataclass, field
+from datetime import datetime
+
+
+@dataclass
+class PollState:
+    portfolio: dict | None = None
+    news_sentiment: dict | None = None
+    screener_preview: dict | None = None
+    minute_bars: dict[str, deque] = field(default_factory=dict)
+    asking_price: dict[str, dict] = field(default_factory=dict)
+    # Phase 3b additions
+    daily_ohlcv: dict[str, list[dict]] = field(default_factory=dict)
+    chronos_predictions: dict[str, dict] = field(default_factory=dict)
+    minute_momentum: dict[str, str] = field(default_factory=dict)
+    signals: dict[str, dict] = field(default_factory=dict)
+    # F5 lifecycle
+    signal_cycle_id: int = 0
+    last_updated: dict[str, str] = field(default_factory=dict)
+    fetch_errors: dict[str, int] = field(default_factory=dict)
+
+    def get_active_signals(self, now: datetime) -> list[dict]:
+        """expires_at > now 인 신호만 반환. expires_at 없으면 expired 취급."""
+        active: list[dict] = []
+        for sig in self.signals.values():
+            expires_at = sig.get("expires_at")
+            if not expires_at:
+                continue
+            try:
+                exp_dt = datetime.fromisoformat(expires_at)
+            except ValueError:
+                continue
+            if exp_dt > now:
+                active.append(sig)
+        return active
+
+    def purge_expired_signals(self, now: datetime) -> int:
+        """만료된 signal 제거. 제거된 개수 반환."""
+        to_drop = []
+        for ticker, sig in self.signals.items():
+            expires_at = sig.get("expires_at")
+            if not expires_at:
+                to_drop.append(ticker)
+                continue
+            try:
+                exp_dt = datetime.fromisoformat(expires_at)
+            except ValueError:
+                to_drop.append(ticker)
+                continue
+            if exp_dt <= now:
+                to_drop.append(ticker)
+        for t in to_drop:
+            del self.signals[t]
+        return len(to_drop)
+
+
+state = PollState()
+```
+
+- [ ] **Step 4: Run test to verify it passes**
+
+```
+python -m pytest ai_trade/tests/test_state_signals_lifecycle.py -v
+```
+
+Expected: 4 PASS.
+
+- [ ] **Step 5: Verify full suite still passes**
+
+```
+python -m pytest ai_trade/tests -q
+```
+
+Expected: 기존 test 전부 PASS (state.signals dict 인터페이스 그대로).
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add ai_trade/state.py ai_trade/tests/test_state_signals_lifecycle.py
+git commit -m "feat(ai_trade): state.signals에 expires_at + cycle_id lifecycle 추가 (F5 part 1)"
+```
+
+---
+
+## Task 2: config에 SIGNAL_TTL_SECONDS 추가
+
+**Files:**
+- Modify: `ai_trade/config.py`
+- Test: `ai_trade/tests/test_state_signals_lifecycle.py` (append)
+
+- [ ] **Step 1: Write failing test**
+
+`test_state_signals_lifecycle.py` 끝에 추가:
+
+```python
+def test_signal_ttl_seconds_default(monkeypatch):
+    monkeypatch.delenv("SIGNAL_TTL_SECONDS", raising=False)
+    from ai_trade.config import Settings
+    s = Settings()
+    assert s.signal_ttl_seconds == 300
+
+
+def test_signal_ttl_seconds_env_override(monkeypatch):
+    monkeypatch.setenv("SIGNAL_TTL_SECONDS", "60")
+    from ai_trade.config import Settings
+    s = Settings()
+    assert s.signal_ttl_seconds == 60
+```
+
+- [ ] **Step 2: Run test to fail**
+
+```
+python -m pytest ai_trade/tests/test_state_signals_lifecycle.py -v -k signal_ttl
+```
+
+Expected: AttributeError.
+
+- [ ] **Step 3: Add field to config.py**
+
+`Settings` 클래스 안에 추가 (다른 *_threshold 옆):
+
+```python
+    signal_ttl_seconds: int = field(
+        default_factory=lambda: int(os.getenv("SIGNAL_TTL_SECONDS", "300"))
+    )
+```
+
+- [ ] **Step 4: Run test**
+
+```
+python -m pytest ai_trade/tests/test_state_signals_lifecycle.py -v -k signal_ttl
+```
+
+Expected: 2 PASS.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add ai_trade/config.py ai_trade/tests/test_state_signals_lifecycle.py
+git commit -m "feat(ai_trade): SIGNAL_TTL_SECONDS env 추가 (F5 part 2)"
+```
+
+---
+
+## Task 3: signal_generator에 cycle_id + expires_at 부착
+
+**Files:**
+- Modify: `ai_trade/signal_generator.py`
+- Test: `ai_trade/tests/test_signal_generator.py` (append)
+
+- [ ] **Step 1: Write failing tests**
+
+기존 `test_signal_generator.py` 끝에 추가:
+
+```python
+from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo as _ZI
+
+_KST_TEST = _ZI("Asia/Seoul")
+
+
+def test_emit_attaches_cycle_id_and_expires_at(
+    state_with_buy_setup, dedup_clean, settings_default,
+):
+    """매 emit 시 cycle_id + expires_at 부착 (F5)."""
+    from ai_trade.signal_generator import generate_signals
+    before = datetime.now(_KST_TEST)
+    generate_signals(state_with_buy_setup, dedup_clean, settings_default)
+    after = datetime.now(_KST_TEST)
+    sig = state_with_buy_setup.signals["005930"]
+    assert sig["cycle_id"] == 1
+    assert "expires_at" in sig
+    exp_dt = datetime.fromisoformat(sig["expires_at"])
+    # as_of + 300s (default) — tolerance 5s
+    assert before + timedelta(seconds=295) < exp_dt < after + timedelta(seconds=305)
+
+
+def test_cycle_id_increments_each_call(
+    state_with_buy_setup, dedup_clean, settings_default,
+):
+    """generate_signals 호출마다 cycle_id += 1."""
+    from ai_trade.signal_generator import generate_signals
+    generate_signals(state_with_buy_setup, dedup_clean, settings_default)
+    assert state_with_buy_setup.signal_cycle_id == 1
+    # 2번째 호출 — dedup이 막아도 cycle_id는 증가해야 함
+    generate_signals(state_with_buy_setup, dedup_clean, settings_default)
+    assert state_with_buy_setup.signal_cycle_id == 2
+```
+
+**NOTE:** 기존 test_signal_generator.py에 `state_with_buy_setup` / `dedup_clean` / `settings_default` 같은 fixture가 있을 것. 만약 이름이 다르면 실제 fixture 이름에 맞춰 조정. 검증: `grep -n "@pytest.fixture" ai_trade/tests/test_signal_generator.py`.
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+```
+python -m pytest ai_trade/tests/test_signal_generator.py -v -k "cycle_id or expires"
+```
+
+Expected: KeyError 또는 AttributeError.
+
+- [ ] **Step 3: Modify signal_generator.py**
+
+`generate_signals` 함수 (L22-25)를 변경:
+
+```python
+def generate_signals(state, dedup, settings) -> None:
+    """Phase 4 entry — state-mutating. F5: cycle_id += 1 + expires_at 부착."""
+    state.signal_cycle_id += 1
+    _evaluate_sell_signals(state, dedup, settings)
+    _evaluate_buy_signals(state, dedup, settings)
+```
+
+`_build_buy_signal` (L99-111)에 두 필드 추가:
+
+```python
+def _build_buy_signal(state, ticker: str, name: str, rank: int | None, confidence: float) -> dict:
+    ap = state.asking_price[ticker]
+    as_of_dt = datetime.now(KST)
+    expires_at = (as_of_dt + timedelta(seconds=getattr(_current_settings(), "signal_ttl_seconds", 300))).isoformat()
+    return {
+        "ticker": ticker,
+        "name": name,
+        "action": "buy",
+        "confidence_webai": confidence,
+        "current_price": ap["current_price"],
+        "avg_price": None,
+        "pnl_pct": None,
+        "context": _build_context(state, ticker, rank),
+        "as_of": as_of_dt.isoformat(),
+        "cycle_id": state.signal_cycle_id,
+        "expires_at": expires_at,
+    }
+```
+
+같이 `_build_sell_signal` (L174-186):
+
+```python
+def _build_sell_signal(state, holding: dict, confidence: float, reason: str, settings=None) -> dict:
+    ticker = holding["ticker"]
+    as_of_dt = datetime.now(KST)
+    ttl = getattr(settings, "signal_ttl_seconds", 300) if settings else 300
+    expires_at = (as_of_dt + timedelta(seconds=ttl)).isoformat()
+    return {
+        "ticker": ticker,
+        "name": holding.get("name", ticker),
+        "action": "sell",
+        "confidence_webai": confidence,
+        "current_price": holding.get("current_price"),
+        "avg_price": holding.get("avg_price"),
+        "pnl_pct": holding.get("pnl_pct"),
+        "context": _build_context(state, ticker, rank=None, sell_reason=reason),
+        "as_of": as_of_dt.isoformat(),
+        "cycle_id": state.signal_cycle_id,
+        "expires_at": expires_at,
+    }
+```
+
+`_build_buy_signal`이 settings를 안 받고 있으니, 호출부도 갱신해야 함. 현실적으로 두 함수에 `settings` 인자를 추가하는 것이 깔끔. 변경:
+
+```python
+def _evaluate_buy_signals(state, dedup, settings) -> None:
+    candidates = _buy_candidates(state)
+    for ticker, name, rank in candidates:
+        existing = state.signals.get(ticker)
+        if existing is not None and existing.get("action") == "sell":
+            logger.debug("buy %s skipped: same-cycle sell precedence", ticker)
+            continue
+        if not _check_buy_hard_gate(state, ticker, settings):
+            logger.debug("buy %s skipped: hard gate failed", ticker)
+            continue
+        confidence = _compute_buy_confidence(state, ticker, rank)
+        if confidence <= settings.confidence_threshold:
+            logger.debug("buy %s skipped: confidence %.3f <= %.3f",
+                         ticker, confidence, settings.confidence_threshold)
+            continue
+        if dedup.is_recent(ticker, "buy", within_hours=24):
+            logger.debug("buy %s skipped: dedup 24h", ticker)
+            continue
+        state.signals[ticker] = _build_buy_signal(state, ticker, name, rank, confidence, settings)
+        dedup.record(ticker, "buy", confidence=confidence)
+        logger.info("signal emit %s buy conf=%.3f rank=%s cycle=%d",
+                    ticker, confidence, rank, state.signal_cycle_id)
+
+
+def _build_buy_signal(state, ticker, name, rank, confidence, settings) -> dict:
+    ap = state.asking_price[ticker]
+    as_of_dt = datetime.now(KST)
+    ttl = settings.signal_ttl_seconds
+    expires_at = (as_of_dt + timedelta(seconds=ttl)).isoformat()
+    return {
+        "ticker": ticker,
+        "name": name,
+        "action": "buy",
+        "confidence_webai": confidence,
+        "current_price": ap["current_price"],
+        "avg_price": None,
+        "pnl_pct": None,
+        "context": _build_context(state, ticker, rank),
+        "as_of": as_of_dt.isoformat(),
+        "cycle_id": state.signal_cycle_id,
+        "expires_at": expires_at,
+    }
+```
+
+매도 측도 마찬가지로 `settings`를 통과시킴. `_try_stop_loss` 등은 이미 `settings`를 받으므로 `_build_sell_signal(..., settings=settings)` 로 호출.
+
+import 추가 (signal_generator.py 상단):
+
+```python
+from datetime import datetime, timedelta
+```
+
+(기존 import에 `timedelta` 만 추가)
+
+`_current_settings()` 같은 헬퍼는 만들지 않음 — settings를 명시적으로 전달.
+
+- [ ] **Step 4: Run tests**
+
+```
+python -m pytest ai_trade/tests/test_signal_generator.py -v
+```
+
+Expected: 신규 2개 PASS, 기존 PASS.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add ai_trade/signal_generator.py ai_trade/tests/test_signal_generator.py
+git commit -m "feat(ai_trade): emit signal에 cycle_id + expires_at 부착 (F5 part 3)"
+```
+
+---
+
+## Task 4: pull_worker가 cycle 끝에 purge 호출
+
+**Files:**
+- Modify: `ai_trade/pull_worker.py`
+- Test: `ai_trade/tests/test_pull_worker.py` (append)
+
+- [ ] **Step 1: Write failing test**
+
+`test_pull_worker.py` 끝에 추가:
+
+```python
+async def test_poll_loop_purges_expired_signals(monkeypatch):
+    """매 cycle 끝에 expired signal이 제거됨 (F5)."""
+    from ai_trade import pull_worker
+    from ai_trade.state import PollState
+    from datetime import datetime as _dt
+    from zoneinfo import ZoneInfo as _ZI
+    from unittest.mock import AsyncMock, MagicMock
+    import asyncio as _asyncio
+
+    _kst = _ZI("Asia/Seoul")
+    now = _dt(2026, 5, 18, 10, 0, tzinfo=_kst)
+
+    class FrozenDT:
+        @staticmethod
+        def now(tz=None): return now
+
+    state = PollState()
+    state.signals = {
+        "OLD": {"ticker": "OLD", "expires_at": _dt(2026, 5, 18, 9, 0, tzinfo=_kst).isoformat(), "cycle_id": 1},
+        "FRESH": {"ticker": "FRESH", "expires_at": _dt(2026, 5, 18, 10, 30, tzinfo=_kst).isoformat(), "cycle_id": 1},
+    }
+
+    monkeypatch.setattr(pull_worker, "datetime", FrozenDT)
+    monkeypatch.setattr(pull_worker, "_is_market_day", lambda n: True)
+    monkeypatch.setattr(pull_worker, "_is_polling_window", lambda n: True)
+    monkeypatch.setattr(pull_worker, "_next_interval", lambda n: 0.01)
+    monkeypatch.setattr(pull_worker, "_run_polling_cycle", AsyncMock())
+    monkeypatch.setattr(pull_worker, "update_minute_momentum_for_all", lambda s: None)
+    monkeypatch.setattr(pull_worker, "_is_post_close_trigger", lambda *a, **k: False)
+
+    shutdown = _asyncio.Event()
+    async def stop_soon():
+        await _asyncio.sleep(0.05)
+        shutdown.set()
+    _asyncio.create_task(stop_soon())
+
+    await pull_worker.poll_loop(
+        client=MagicMock(), state=state, shutdown=shutdown,
+        kis_client=MagicMock(), chronos=MagicMock(),
+        dedup=None, settings=None,
+    )
+    assert "OLD" not in state.signals
+    assert "FRESH" in state.signals
+```
+
+- [ ] **Step 2: Run test to fail**
+
+```
+python -m pytest ai_trade/tests/test_pull_worker.py::test_poll_loop_purges_expired_signals -v
+```
+
+Expected: FAIL — OLD가 남아있음.
+
+- [ ] **Step 3: Add purge call in poll_loop**
+
+`ai_trade/pull_worker.py` `poll_loop` 안, signals 생성 이후 (또는 cycle 끝 직전) 한 줄 추가:
+
+```python
+            # Phase 4: generate signals
+            if dedup is not None and settings is not None:
+                try:
+                    from ai_trade.signal_generator import generate_signals
+                    generate_signals(state, dedup, settings)
+                except Exception:
+                    logger.exception("generate_signals failed")
+            # F5: 만료된 signal purge (consumer 미사용 케이스 보호)
+            try:
+                state.purge_expired_signals(datetime.now(KST))
+            except Exception:
+                logger.exception("purge_expired_signals failed")
+```
+
+- [ ] **Step 4: Run test**
+
+```
+python -m pytest ai_trade/tests/test_pull_worker.py::test_poll_loop_purges_expired_signals -v
+```
+
+Expected: PASS.
+
+- [ ] **Step 5: Run full suite**
+
+```
+python -m pytest ai_trade/tests -q
+```
+
+Expected: 모두 PASS.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add ai_trade/pull_worker.py ai_trade/tests/test_pull_worker.py
+git commit -m "feat(ai_trade): poll_loop가 매 cycle 끝에 expired signal purge (F5 part 4)"
+```
+
+---
+
+## Task 5: 전체 회귀 + push
+
+- [ ] **Step 1: Final pytest**
+
+```
+python -m pytest ai_trade/tests -v
+```
+
+Expected: 모두 PASS (총 신규 약 9개 + 기존 56개).
+
+- [ ] **Step 2: Push**
+
+```bash
+git push origin main
+```
+
+---
+
+## Self-Review
+
+1. **expires_at + cycle_id 부착**: `_build_buy_signal`, `_build_sell_signal` 양쪽 ✅
+2. **cycle_id 증가**: `generate_signals` 진입에서 단 1회 ✅
+3. **purge**: poll_loop cycle 마지막에 1회 호출 ✅
+4. **get_active_signals**: Phase 5 consumer가 호출할 read API 존재 ✅
+5. **legacy 신호 호환**: `expires_at` 없는 신호는 expired 취급 → 안전 ✅
+
+**미커버 항목 (의도적)**:
+- Phase 5 consumer가 처리 후 explicit drain하는 API는 이 plan에서 안 다룸 (consumer가 read-only로도 충분 — expires_at + dedup으로 idempotent).
+- agent-office `/signal` HTTP endpoint는 Phase 5 plan 영역.
+
+---
+
+## Execution Handoff
+
+**Plan complete and saved to `docs/superpowers/plans/2026-05-25-state-signals-lifecycle.md`.**
+
+**1. Subagent-Driven (recommended)** — Task 별 fresh subagent.
+**2. Inline Execution** — 현 세션 실행.
+
+박재오 결정 대기. Plan 1 (hotfix) 마친 뒤 진입 권장.

docs/superpowers/specs/2026-05-15-confidence-signal-pipeline-v2-architecture.md

@@ -194,7 +194,7 @@ agent-office 가 web-ai 의 Ollama (Qwen3 14B Q4) 에 보내는 prompt 의 응
 ### 6.1 매수 신호 (screener Top-20 종목 대상)
 
 조건 (전부 충족):
-1. Chronos-2 1-day quantile (median) 예측 > 0% 그리고 분포 폭 (90-10 분위수 / 50 분위수) < 0.6 (좁은 분포 = 높은 conf)
+1. Chronos-2 1-day quantile (median) 예측 > 0% 그리고 분포 폭 `q90 - q10` < 0.6 (절대 spread, 60% return 변동 미만 = 모델 확신; **Phase 4 amend 2026-05-17**: 기존 relative formula `(q90-q10)/median` 는 Chronos-bolt 의 median≈0 출력에서 거의 모든 신호 거부 → absolute spread 채택. 자세한 사유는 `2026-05-17-signal-v2-phase4-signal-generator.md` §4.2 참조)
 2. 분봉 모멘텀 = `strong_up`:
    - 5분봉 5개 연속 양봉
    - 거래량 > 평균 1.5배

docs/superpowers/specs/2026-05-17-agent-office-grid-redesign-design.md

@@ -0,0 +1,345 @@
+# Agent Office 그리드 재설계 — Design Spec
+
+**Date:** 2026-05-17
+**Author:** CEO (with Claude)
+**Target:** `web-ui` `/agent-office` 페이지
+
+---
+
+## 1. 배경 & 목적
+
+현재 `/agent-office` 페이지는 픽셀 사무실 Canvas 위에서 5명의 에이전트 캐릭터가 무의미하게 걸어다니는 형태다. 시각적 즐거움은 있으나 정보 밀도가 낮고, 각 에이전트가 무슨 일을 하는지 한눈에 파악하기 어렵다.
+
+이를 **3x3 그리드** 기반의 정보 중심 UI로 재설계한다. 왼편에 9개의 에이전트 이미지 카드를 배치하고, 카드 클릭 시 오른편 패널에서 해당 에이전트의 명령·태스크·토큰·로그를 확인한다.
+
+---
+
+## 2. 범위 (Scope)
+
+### In scope
+- `src/pages/agent-office/AgentOffice.jsx` 전면 재작성 (Canvas → Grid)
+- 그리드 카드 컴포넌트 신규 작성
+- `SidePanel.jsx` 헤더 부분 수정 (emoji → 이미지)
+- `SidePanel.jsx`의 `AGENT_META`에서 `blog` 제거, `insta` 추가
+- TopBar 단순화 (theme/zoom 컨트롤 제거)
+- Canvas 관련 파일/디렉토리 전체 삭제
+- 이미지 에셋 디렉토리 신설
+
+### Out of scope
+- 백엔드 변경 (현재 백엔드의 `insta` 에이전트는 이미 등록 완료, 추가 작업 불필요)
+- 새 에이전트 추가 (4개 placeholder는 "준비 중" 표시만)
+- 4탭 컨텐츠 (Commands/Tasks/Tokens/Logs) 로직 수정
+
+---
+
+## 3. 에이전트 구성
+
+### 실제 작동 5명
+
+| ID | 표시명 | 색상 | 역할 요약 |
+|----|--------|------|-----------|
+| `stock` | 주식 트레이더 | `#4488cc` | 주식 매매·뉴스 분석·포트폴리오 |
+| `music` | 음악 프로듀서 | `#44aa88` | AI 음악 생성 |
+| `insta` | 인스타 큐레이터 | `#d97706` | 매일 09:30 뉴스 수집 → 키워드 추출 → AI 카드 10장 생성 → 텔레그램 푸시 |
+| `realestate` | 청약 애널리스트 | `#c026d3` | 부동산 청약 매칭·자치구 5티어 분석 |
+| `lotto` | 로또 큐레이터 | `#ef4444` | 로또 번호 추천·브리핑 |
+
+> `blog`는 `insta`로 대체됨. 기존 `SidePanel.jsx`의 `AGENT_META`에서 `blog` 항목 삭제 + `insta` 추가.
+
+### Placeholder 4개
+
+- ID 없음 (그리드 슬롯 인덱스 6/7/8/9로만 식별)
+- 모두 동일하게 `agent_undetermined.png` + "준비 중" 라벨
+- 클릭 시 정적 안내 패널 노출
+
+---
+
+## 4. 디렉토리 & 파일 구조
+
+### 신설 디렉토리
+
+```
+src/pages/agent-office/assets/agents/
+├── agent_stock.png          (사용자 제공)
+├── agent_music.png          (사용자 제공)
+├── agent_insta.png          (사용자 제공)
+├── agent_realestate.png     (사용자 제공)
+├── agent_lotto.png          (사용자 제공)
+└── agent_undetermined.png   (사용자 제공, 4 placeholder 공유)
+```
+
+### 파일명 규칙
+`agent_{id}.png` 형식. `{id}`는 백엔드의 agent_id와 일치 (소문자, underscore).
+
+### 권장 이미지 사양
+- 정사각형 (예: 512x512)
+- PNG (투명 배경 허용)
+- 카드 표시 시 `object-fit: cover`로 정사각 크롭
+
+### 삭제 대상
+
+```
+src/pages/agent-office/
+├── canvas/                  ← 전체 삭제
+│   ├── themes.js
+│   ├── FurnitureRenderer.js
+│   ├── ProceduralSprite.js
+│   ├── AgentSprite.js
+│   ├── SpriteLoader.js
+│   ├── OverlayRenderer.js
+│   ├── Pathfinder.js
+│   ├── OfficeRenderer.js
+│   └── TileMap.js
+├── hooks/
+│   └── useOfficeCanvas.js   ← 삭제
+└── assets/
+    └── office-map.json      ← 삭제
+```
+
+### 유지 대상
+
+```
+src/pages/agent-office/
+├── AgentOffice.jsx               ← 재작성
+├── AgentOffice.css               ← 재작성
+├── hooks/
+│   └── useAgentManager.js        ← 그대로 (WebSocket 로직)
+└── components/
+    ├── TopBar.jsx                ← 단순화 (theme/zoom 제거)
+    ├── SidePanel.jsx             ← 헤더 수정 + AGENT_META 갱신
+    ├── CommandTab.jsx            ← 그대로
+    ├── TaskTab.jsx               ← 그대로
+    ├── TokenTab.jsx              ← 그대로
+    └── LogTab.jsx                ← 그대로
+```
+
+### 신규 컴포넌트
+
+```
+src/pages/agent-office/components/
+├── AgentGrid.jsx           ← 3x3 그리드 래퍼
+├── AgentCard.jsx           ← 카드 1개 (image + state dot + badge + name)
+├── PlaceholderCard.jsx     ← "준비 중" 카드
+└── EmptyDetailPanel.jsx    ← 초기 안내 / placeholder 클릭 시 안내
+```
+
+---
+
+## 5. 레이아웃
+
+### 전체 화면 구조
+
+```
+┌─────────────────────────────────────────────────────────────┐
+│  TopBar (connected status only)                              │
+├──────────────────────────────────┬──────────────────────────┤
+│                                  │                          │
+│  AgentGrid (3x3)                 │  Right Panel              │
+│  ┌──────┬──────┬──────┐          │                          │
+│  │stock │music │insta │          │  ┌─ active 선택 시 ─┐    │
+│  ├──────┼──────┼──────┤          │  │ SidePanel        │    │
+│  │realE │lotto │  ??  │          │  │ - 헤더(이미지+이름)│    │
+│  ├──────┼──────┼──────┤          │  │ - 4 tabs         │    │
+│  │  ??  │  ??  │  ??  │          │  └──────────────────┘    │
+│  └──────┴──────┴──────┘          │                          │
+│                                  │  ┌─ placeholder 선택 ─┐  │
+│                                  │  │ "준비 중인 에이전트"│  │
+│                                  │  └────────────────────┘  │
+│                                  │                          │
+│                                  │  ┌─ 초기(미선택) ──────┐ │
+│                                  │  │ "에이전트를 선택…"  │ │
+│                                  │  └────────────────────┘ │
+└──────────────────────────────────┴──────────────────────────┘
+```
+
+### 그리드 슬롯 순서 (좌→우, 위→아래)
+
+| Index | Slot |
+|-------|------|
+| 1 (행1·열1) | `stock` |
+| 2 (행1·열2) | `music` |
+| 3 (행1·열3) | `insta` |
+| 4 (행2·열1) | `realestate` |
+| 5 (행2·열2) | `lotto` |
+| 6 (행2·열3) | placeholder |
+| 7 (행3·열1) | placeholder |
+| 8 (행3·열2) | placeholder |
+| 9 (행3·열3) | placeholder |
+
+### AgentCard 시각 구조
+
+```
+┌─────────────────────┐
+│ ● state    [③]      │ ← 상태 dot(좌상, image 약간 위) + 알림 뱃지(우상)
+│  ┌───────────────┐  │
+│  │               │  │
+│  │  agent_xxx    │  │ ← 정사각 이미지 (object-fit: cover)
+│  │     .png      │  │
+│  │               │  │
+│  └───────────────┘  │
+│   주식 트레이더      │ ← display_name
+└─────────────────────┘
+```
+
+#### 상태 dot
+
+| state | color | 동작 |
+|-------|-------|------|
+| `idle` | `#6b7280` (회색) | 정적 |
+| `working` | `#22c55e` (초록) | pulse 애니메이션 |
+| `error` | `#ef4444` (빨강) | 정적 |
+| `waiting_approval` | `#f59e0b` (주황) | pulse |
+| `break` | `#94a3b8` (밝은 회색) | 정적 |
+
+상태 dot은 카드의 좌상단, 이미지보다 약간 위쪽에 위치 (이미지 영역 바깥 또는 모서리 살짝 걸침).
+
+#### 알림 뱃지
+
+- `notifications[agentId] > 0`일 때만 우상단에 표시
+- 빨강 배경에 흰 숫자 (count > 9면 "9+")
+- 카드 클릭 시 자동으로 0으로 리셋 (`clearNotifications` 호출 — 기존 로직 재사용)
+
+---
+
+## 6. 데이터 플로우
+
+```
+useAgentManager (그대로 유지)
+   ├── WebSocket /api/agent-office/ws
+   ├── agents: { [id]: { state, detail, task_id } }
+   ├── notifications: { [id]: count }
+   ├── pendingTasks: [...]
+   ├── connected: bool
+   └── refreshTrigger: number
+
+AgentOffice.jsx
+   ├── agents → AgentGrid에 전달 → 각 AgentCard가 state로 dot 색상 결정
+   ├── notifications → 각 AgentCard가 badge 표시
+   ├── selectedAgent (local state): string | null | "placeholder"
+   └── 카드 클릭 시 setSelectedAgent + clearNotifications
+
+Right Panel 분기
+   ├── selectedAgent === null         → EmptyDetailPanel (초기 안내)
+   ├── selectedAgent === "placeholder"→ EmptyDetailPanel ("준비 중" 메시지)
+   └── selectedAgent ∈ active 5명     → SidePanel (4탭, 기존 로직)
+```
+
+---
+
+## 7. SidePanel 수정 사항
+
+### AGENT_META 갱신
+
+```js
+// src/pages/agent-office/components/SidePanel.jsx
+import stockImg from '../assets/agents/agent_stock.png';
+import musicImg from '../assets/agents/agent_music.png';
+import instaImg from '../assets/agents/agent_insta.png';
+import realestateImg from '../assets/agents/agent_realestate.png';
+import lottoImg from '../assets/agents/agent_lotto.png';
+
+const AGENT_META = {
+  stock:      { displayName: '주식 트레이더',  image: stockImg,      color: '#4488cc' },
+  music:      { displayName: '음악 프로듀서',  image: musicImg,      color: '#44aa88' },
+  insta:      { displayName: '인스타 큐레이터', image: instaImg,      color: '#d97706' },
+  realestate: { displayName: '청약 애널리스트', image: realestateImg, color: '#c026d3' },
+  lotto:      { displayName: '로또 큐레이터',  image: lottoImg,      color: '#ef4444' }
+};
+// blog 항목 삭제
+```
+
+### 헤더 시각 변경
+
+```jsx
+// 변경 전: emoji icon
+<div className="ao-sidepanel-icon" style={{ background: meta.color }}>
+  {meta.emoji}
+</div>
+
+// 변경 후: 이미지
+<div className="ao-sidepanel-icon" style={{ borderColor: meta.color }}>
+  <img src={meta.image} alt={meta.displayName} />
+</div>
+```
+
+4탭(Commands/Tasks/Tokens/Logs) 본체 로직은 손대지 않음.
+
+---
+
+## 8. CSS 토큰 (제안)
+
+```css
+:root {
+  --ao-bg: #0f172a;
+  --ao-card-bg: #1e293b;
+  --ao-card-border: #334155;
+  --ao-card-border-active: #60a5fa;
+  --ao-text: #e2e8f0;
+  --ao-text-muted: #94a3b8;
+  --ao-grid-gap: 16px;
+  --ao-card-radius: 12px;
+}
+
+.ao-grid {
+  display: grid;
+  grid-template-columns: repeat(3, 1fr);
+  gap: var(--ao-grid-gap);
+}
+
+.ao-card {
+  aspect-ratio: 1 / 1.15;  /* 이미지 정사각 + 이름줄 */
+  background: var(--ao-card-bg);
+  border: 1px solid var(--ao-card-border);
+  border-radius: var(--ao-card-radius);
+  cursor: pointer;
+  transition: transform 120ms ease, border-color 120ms ease;
+}
+
+.ao-card:hover { transform: translateY(-2px); }
+.ao-card.active { border-color: var(--ao-card-border-active); }
+.ao-card.placeholder { opacity: 0.55; cursor: pointer; }
+```
+
+반응형: 모바일에서는 `grid-template-columns: repeat(2, 1fr)` 또는 `repeat(1, 1fr)`로 축소.
+
+---
+
+## 9. 에러 처리 / Edge Cases
+
+| 상황 | 동작 |
+|------|------|
+| 이미지 로드 실패 | `<img onError>`로 단색 배경 + 첫 글자 fallback |
+| WebSocket 끊김 | TopBar에 disconnected 표시. 카드는 마지막 상태 유지 (회색 처리 안 함 — 기존 동작 유지) |
+| `agents[id]` 미존재 | dot 회색(`idle`), 정상 표시 |
+| placeholder 클릭 | 우측 패널만 변경, WebSocket 호출/clearNotifications 호출 없음 |
+
+---
+
+## 10. 테스트 계획
+
+- [ ] 6개 이미지 파일이 디렉토리에 존재할 때 그리드 정상 렌더링
+- [ ] 이미지 누락 시 fallback 표시
+- [ ] WebSocket으로 `agent_state` 수신 시 dot 색상 변경
+- [ ] `notification` 수신 시 뱃지 표시, 카드 클릭 시 0으로 리셋
+- [ ] active 5명 클릭 → SidePanel 4탭 표시 (기존 동작 유지)
+- [ ] placeholder 4슬롯 클릭 → "준비 중" 패널
+- [ ] TopBar의 connected/disconnected 표시 정상
+- [ ] Canvas 잔재(파일 import 누락 등) 없음 — `npm run build` 통과
+- [ ] 모바일 뷰(<768px) 그리드 축소 정상
+
+---
+
+## 11. 이행 절차 (사용자 작업 포함)
+
+1. **사용자**: `src/pages/agent-office/assets/agents/` 디렉토리에 6개 PNG 파일 배치
+2. **Claude (구현 단계)**: writing-plans 스킬로 단계별 작업 계획 작성
+3. 구현·삭제·테스트 후 commit
+4. NAS 배포는 별도 (`npm run release:nas`)
+
+---
+
+## 12. 향후 확장
+
+- 9번째 active 에이전트 채용 시: 이미지 추가 + `AGENT_META` 갱신 + 슬롯 인덱스 매핑 변경
+- 그리드 자동 정렬(상태별/우선순위별 sort) — 현재는 정적 배치
+- 카드 hover 시 미니 프리뷰 (최근 활동 1줄 요약) — 추후 검토

docs/superpowers/specs/2026-05-17-signal-v2-phase4-signal-generator.md

@@ -118,10 +118,16 @@ Phase 2/3a/3b 의 모든 산출을 종합해 Phase 0 spec §6.1/§6.2/§6.3 의
 ### 4.2 Hard gate (모든 조건 충족)
 
 1. `state.chronos_predictions[ticker].median > 0` (다음날 상승)
-2. `(q90 - q10) / max(|median|, 0.001) < settings.chronos_spread_threshold`
+2. `(q90 - q10) < settings.chronos_spread_threshold` (**absolute spread** — Phase 3b 실 운영 데이터 기반 변경)
 3. `state.minute_momentum[ticker] == settings.min_momentum_for_buy` (기본 strong_up)
 4. `state.asking_price[ticker].bid_ratio >= settings.asking_bid_ratio_threshold`
 
+**Spread formula 결정 노트 (2026-05-17 implementer 변경 채택)**:
+- Phase 0 spec §6.1 의 한국어 "(90-10 분위수) / 50 분위수 < 0.6" 은 *relative spread* 로 명시되었으나, Phase 3b 실 운영 결과 (Chronos zero-shot prediction 의 median 이 종종 0 가까이) 에서 relative formula 가 거의 모든 신호 거부 → useless.
+- **변경**: absolute spread `(q90 - q10) < 0.6` 사용. 0.6 = 60% 변동 예측 — 한국 주식 1-day 변동성 (1-2%) 대비 매우 넓음 (모델 자신 없음 신호).
+- 결과: Phase 3b smoke 005930 (median=-0.59%, q10=-8.9%, q90=6.4%, spread=15.3%) → spread 0.153 < 0.6 → hard gate 통과 가능 (다른 조건 충족 시).
+- Phase 7 IC 검증 시 임계값 재조정 가능 (env `CHRONOS_SPREAD_THRESHOLD`).
+
 ### 4.3 Soft confidence (Phase 0 spec §6.1)
 
 ```python

index.html

@@ -5,6 +5,13 @@
     <link rel="icon" type="image/svg+xml" href="/main_logo.png" />
     <meta name="viewport" content="width=device-width, initial-scale=1.0, viewport-fit=cover" />
     <title>가후습 개인기록</title>
+    <link rel="preconnect" href="https://fonts.googleapis.com" />
+    <link rel="preconnect" href="https://fonts.gstatic.com" crossorigin />
+    <link href="https://fonts.googleapis.com/css2?family=Noto+Serif+KR:wght@500;700&display=swap" rel="stylesheet" />
+    <link
+      href="https://fonts.googleapis.com/css2?family=Nanum+Myeongjo:wght@400;700;800&family=Nanum+Gothic:wght@400;700;800&family=Gowun+Batang:wght@400;700&display=swap"
+      rel="stylesheet"
+    />
   </head>
   <body>
     <div id="root"></div>

public/images/tarot/card_back.svg

@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 200 300" width="200" height="300">
+  <defs>
+    <linearGradient id="bg" x1="0" y1="0" x2="0" y2="1">
+      <stop offset="0%" stop-color="#1a0d2e"/>
+      <stop offset="100%" stop-color="#0a0420"/>
+    </linearGradient>
+    <linearGradient id="goldFrame" x1="0" y1="0" x2="1" y2="1">
+      <stop offset="0%" stop-color="#d4af37"/>
+      <stop offset="100%" stop-color="#8b6914"/>
+    </linearGradient>
+  </defs>
+  <rect width="200" height="300" rx="14" fill="url(#bg)"/>
+  <rect x="8" y="8" width="184" height="284" rx="10" fill="none"
+        stroke="url(#goldFrame)" stroke-width="2"/>
+  <g transform="translate(100 150)" fill="#d4af37" font-family="serif" text-anchor="middle">
+    <circle r="38" fill="none" stroke="#d4af37" stroke-width="1.5"/>
+    <text font-size="48" dy="14" font-style="italic">A</text>
+    <g opacity=".5">
+      <circle cx="-60" cy="-90" r="1.5"/>
+      <circle cx="55" cy="-100" r="1"/>
+      <circle cx="-50" cy="80" r="1.2"/>
+      <circle cx="65" cy="90" r="1"/>
+      <circle cx="0" cy="-110" r="1.6"/>
+    </g>
+  </g>
+  <text x="100" y="280" fill="#d4af37" font-family="serif" font-size="10"
+        text-anchor="middle" letter-spacing="2">ARCANA TAROT</text>
+</svg>