fix(tarot): Claude 응답 시간 단축 + nginx timeout 정리

504 Gateway Timeout 근본 원인은 DSM Reverse Proxy의 60s 기본 timeout
(agent-office는 200 OK 정상 응답했으나 client 도달 전 DSM이 끊음).
사용자 측 DSM Reverse Proxy timeout 늘리기 별도 필요.

코드 측 대응:
- pipeline.py max_tokens 2048 → 1400 (응답 시간 단축, 3-card spread 충분)
- pipeline.py에 latency_ms·tokens 로그 출력 (모니터링)
- nginx /api/agent-office/에 proxy_send_timeout 300s, proxy_connect_timeout 60s
  추가 (proxy_read_timeout은 WebSocket 위해 86400s 유지)

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

agent-office/app/agents/insta.py

@@ -18,6 +18,26 @@ from ..telegram import messaging
 
 logger = logging.getLogger(__name__)
 
+# 텔레그램 후보 푸시 시 "확실한 것만" 보내기 위한 최소 신뢰도 (키워드 score 0~1)
+KEYWORD_MIN_SCORE = 0.7
+
+
+def _dedup_and_filter_keywords(
+    keywords: List[Dict[str, Any]], min_score: float = KEYWORD_MIN_SCORE,
+) -> List[Dict[str, Any]]:
+    """score >= min_score 인 키워드만 남기고, 동일 keyword 중복 제거(최고 score 유지).
+    결과는 score 내림차순. 텔레그램 후보 푸시 전 정리용."""
+    best: Dict[str, Dict[str, Any]] = {}
+    for k in keywords:
+        if float(k.get("score", 0)) < min_score:
+            continue
+        name = str(k.get("keyword", "")).strip()
+        if not name:
+            continue
+        if name not in best or k["score"] > best[name]["score"]:
+            best[name] = k
+    return sorted(best.values(), key=lambda k: -k["score"])
+
 
 async def _send_media_group(media: List[Dict[str, Any]], caption: str = "") -> Dict[str, Any]:
     """텔레그램 sendMediaGroup. media는 InputMediaPhoto dicts.
@@ -89,14 +109,18 @@ class InstaAgent(BaseAgent):
         raise TimeoutError(f"{step} timeout {timeout_sec}s")
 
     async def _push_keyword_candidates(self, keywords: List[Dict[str, Any]]) -> None:
-        by_cat: Dict[str, List[Dict[str, Any]]] = {}
+        # 중복 제거 + 신뢰도(score) 임계값 이상만 — "확실한 것만" 정리해서 전송
-        for k in keywords:
+        filtered = _dedup_and_filter_keywords(keywords)
-            by_cat.setdefault(k["category"], []).append(k)
+        if not filtered:
-        if not by_cat:
+            await messaging.send_raw(
-            await messaging.send_raw("📰 [인스타 큐레이터] 오늘은 추천할 키워드가 없습니다.")
+                f"📰 [인스타 큐레이터] 오늘은 확실한 추천 키워드가 없습니다 (신뢰도 {KEYWORD_MIN_SCORE:.1f}+ 기준)."
+            )
             return
+        by_cat: Dict[str, List[Dict[str, Any]]] = {}
+        for k in filtered:
+            by_cat.setdefault(k["category"], []).append(k)
         rows: List[List[Dict[str, Any]]] = []
-        text_lines = ["📰 <b>[인스타 큐레이터]</b> 오늘의 키워드 후보"]
+        text_lines = [f"📰 <b>[인스타 큐레이터]</b> 오늘의 키워드 후보 (신뢰도 {KEYWORD_MIN_SCORE:.1f}+)"]
         for cat, items in by_cat.items():
             text_lines.append(f"\n<b>{cat}</b>")
             for k in items[:5]:

agent-office/app/config.py

@@ -38,3 +38,9 @@ LOTTO_DIGEST_HOUR = int(os.getenv("LOTTO_DIGEST_HOUR", "9"))
 LOTTO_DIGEST_MIN = int(os.getenv("LOTTO_DIGEST_MIN", "25"))
 LOTTO_THROTTLE_HOURS = int(os.getenv("LOTTO_THROTTLE_HOURS", "6"))
 LOTTO_URGENT_DAILY_MAX = int(os.getenv("LOTTO_URGENT_DAILY_MAX", "3"))
+
+# Tarot Lab
+TAROT_MODEL = os.getenv("TAROT_MODEL", "claude-sonnet-4-6")
+TAROT_COST_INPUT_PER_M = float(os.getenv("TAROT_COST_INPUT_PER_M", "3.0"))
+TAROT_COST_OUTPUT_PER_M = float(os.getenv("TAROT_COST_OUTPUT_PER_M", "15.0"))
+TAROT_TIMEOUT_SEC = int(os.getenv("TAROT_TIMEOUT_SEC", "60"))

agent-office/app/db.py

@@ -131,6 +131,33 @@ def init_db() -> None:
                 updated_at          TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
             )
         """)
+        conn.execute("""
+            CREATE TABLE IF NOT EXISTS tarot_readings (
+                id                  INTEGER PRIMARY KEY AUTOINCREMENT,
+                created_at          TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
+                spread_type         TEXT NOT NULL,
+                category            TEXT,
+                question            TEXT,
+                cards               TEXT NOT NULL,
+                interpretation_json TEXT,
+                summary             TEXT,
+                model               TEXT,
+                tokens_in           INTEGER,
+                tokens_out          INTEGER,
+                cost_usd            REAL,
+                confidence          TEXT,
+                favorite            INTEGER NOT NULL DEFAULT 0,
+                note                TEXT
+            )
+        """)
+        conn.execute("""
+            CREATE INDEX IF NOT EXISTS idx_tarot_created
+            ON tarot_readings(created_at DESC)
+        """)
+        conn.execute("""
+            CREATE INDEX IF NOT EXISTS idx_tarot_favorite
+            ON tarot_readings(favorite, created_at DESC)
+        """)
         # Seed default agent configs
         for agent_id, name in [
             ("stock", "주식 트레이더"),
@@ -766,3 +793,117 @@ def get_tasks_by_agent_date_kind(agent_id: str, date_iso: str, task_type: str) -
             (agent_id, task_type, date_iso),
         ).fetchall()
     return [_task_to_dict(r) for r in rows]
+
+
+# --- tarot_readings CRUD ---
+
+def save_tarot_reading(data: Dict[str, Any]) -> int:
+    interp = data.get("interpretation_json") or {}
+    summary = interp.get("summary", "") if isinstance(interp, dict) else ""
+    with _conn() as conn:
+        cur = conn.execute(
+            """INSERT INTO tarot_readings
+               (spread_type, category, question, cards, interpretation_json,
+                summary, model, tokens_in, tokens_out, cost_usd, confidence)
+               VALUES (?,?,?,?,?,?,?,?,?,?,?)""",
+            (
+                data["spread_type"],
+                data.get("category"),
+                data.get("question"),
+                json.dumps(data.get("cards") or [], ensure_ascii=False),
+                json.dumps(interp, ensure_ascii=False) if interp else None,
+                summary,
+                data.get("model"),
+                data.get("tokens_in"),
+                data.get("tokens_out"),
+                data.get("cost_usd"),
+                data.get("confidence"),
+            ),
+        )
+        return int(cur.lastrowid)
+
+
+def get_tarot_reading(reading_id: int) -> Optional[Dict[str, Any]]:
+    with _conn() as conn:
+        r = conn.execute("SELECT * FROM tarot_readings WHERE id=?", (reading_id,)).fetchone()
+    return _tarot_row_to_dict(r) if r else None
+
+
+def list_tarot_readings(
+    page: int = 1, size: int = 20,
+    favorite: Optional[bool] = None,
+    spread_type: Optional[str] = None,
+    category: Optional[str] = None,
+) -> Dict[str, Any]:
+    wheres, params = [], []
+    if favorite is not None:
+        wheres.append("favorite=?")
+        params.append(1 if favorite else 0)
+    if spread_type:
+        wheres.append("spread_type=?")
+        params.append(spread_type)
+    if category:
+        wheres.append("category=?")
+        params.append(category)
+    where_sql = ("WHERE " + " AND ".join(wheres)) if wheres else ""
+    offset = (page - 1) * size
+    with _conn() as conn:
+        total = conn.execute(
+            f"SELECT COUNT(*) c FROM tarot_readings {where_sql}", params
+        ).fetchone()["c"]
+        rows = conn.execute(
+            f"SELECT * FROM tarot_readings {where_sql} ORDER BY created_at DESC LIMIT ? OFFSET ?",
+            params + [size, offset],
+        ).fetchall()
+    return {
+        "items": [_tarot_row_to_dict(r) for r in rows],
+        "page": page, "size": size, "total": int(total),
+    }
+
+
+def update_tarot_reading(reading_id: int, **kwargs) -> None:
+    sets, vals = [], []
+    if "favorite" in kwargs and kwargs["favorite"] is not None:
+        sets.append("favorite=?")
+        vals.append(1 if kwargs["favorite"] else 0)
+    if "note" in kwargs and kwargs["note"] is not None:
+        sets.append("note=?")
+        vals.append(kwargs["note"])
+    if not sets:
+        return
+    vals.append(reading_id)
+    with _conn() as conn:
+        conn.execute(f"UPDATE tarot_readings SET {','.join(sets)} WHERE id=?", vals)
+
+
+def delete_tarot_reading(reading_id: int) -> None:
+    with _conn() as conn:
+        conn.execute("DELETE FROM tarot_readings WHERE id=?", (reading_id,))
+
+
+def _tarot_row_to_dict(r) -> Dict[str, Any]:
+    try:
+        interp = json.loads(r["interpretation_json"]) if r["interpretation_json"] else None
+    except (ValueError, TypeError):
+        interp = None
+    try:
+        cards = json.loads(r["cards"]) if r["cards"] else []
+    except (ValueError, TypeError):
+        cards = []
+    return {
+        "id": r["id"],
+        "created_at": r["created_at"],
+        "spread_type": r["spread_type"],
+        "category": r["category"],
+        "question": r["question"],
+        "cards": cards,
+        "interpretation_json": interp,
+        "summary": r["summary"],
+        "model": r["model"],
+        "tokens_in": r["tokens_in"],
+        "tokens_out": r["tokens_out"],
+        "cost_usd": r["cost_usd"],
+        "confidence": r["confidence"],
+        "favorite": int(r["favorite"]),
+        "note": r["note"],
+    }

agent-office/app/main.py

@@ -12,9 +12,11 @@ from .agents import init_agents, get_agent, get_all_agent_states, AGENT_REGISTRY
 from .scheduler import init_scheduler
 from . import telegram_bot
 from .routers import notify as notify_router
+from .routers import tarot as tarot_router
 
 app = FastAPI()
 app.include_router(notify_router.router)
+app.include_router(tarot_router.router)
 
 _cors_origins = CORS_ALLOW_ORIGINS.split(",")
 app.add_middleware(

agent-office/app/models.py

@@ -1,5 +1,5 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
-from typing import Optional
+from typing import Optional, List, Literal
 
 
 class CommandRequest(BaseModel):
@@ -33,3 +33,46 @@ class ComposeCommand(BaseModel):
     style: Optional[str] = None
     model: Optional[str] = "V4"
     instrumental: Optional[bool] = False
+
+
+class TarotCardDraw(BaseModel):
+    position: str
+    card_id: str
+    reversed: bool = False
+
+
+class TarotInterpretRequest(BaseModel):
+    spread_type: Literal["one_card", "three_card"]
+    category: Optional[str] = None
+    question: Optional[str] = None
+    cards: List[TarotCardDraw]
+    cards_reference: str = Field(..., min_length=1)
+    context_meta: dict = Field(default_factory=dict)
+
+
+class TarotInterpretResponse(BaseModel):
+    interpretation_json: dict
+    model: str
+    tokens_in: int
+    tokens_out: int
+    cost_usd: float
+    latency_ms: int
+    reroll_count: int = 0
+
+
+class TarotSaveRequest(BaseModel):
+    spread_type: Literal["one_card", "three_card"]
+    category: Optional[str] = None
+    question: Optional[str] = None
+    cards: List[TarotCardDraw]
+    interpretation_json: dict
+    model: str
+    tokens_in: int
+    tokens_out: int
+    cost_usd: float
+    confidence: Optional[str] = None
+
+
+class TarotPatchRequest(BaseModel):
+    favorite: Optional[bool] = None
+    note: Optional[str] = None

agent-office/app/routers/tarot.py

@@ -0,0 +1,70 @@
+"""Tarot Lab 엔드포인트 — interpret + readings CRUD."""
+from fastapi import APIRouter, HTTPException
+
+from ..models import (
+    TarotInterpretRequest,
+    TarotInterpretResponse,
+    TarotSaveRequest,
+    TarotPatchRequest,
+)
+from ..tarot import pipeline
+from .. import db as db_module
+
+
+router = APIRouter(prefix="/api/agent-office/tarot")
+
+
+@router.post("/interpret", response_model=TarotInterpretResponse)
+async def interpret_endpoint(req: TarotInterpretRequest):
+    try:
+        result = await pipeline.interpret(req)
+    except pipeline.TarotError as e:
+        raise HTTPException(status_code=500, detail=str(e)) from e
+    return result
+
+
+@router.post("/readings")
+async def save_reading(req: TarotSaveRequest):
+    rid = db_module.save_tarot_reading(req.model_dump())
+    row = db_module.get_tarot_reading(rid)
+    return {"id": rid, "created_at": row["created_at"]}
+
+
+@router.get("/readings")
+async def list_readings(
+    page: int = 1,
+    size: int = 20,
+    favorite: bool | None = None,
+    spread_type: str | None = None,
+    category: str | None = None,
+):
+    return db_module.list_tarot_readings(
+        page=page, size=size,
+        favorite=favorite, spread_type=spread_type, category=category,
+    )
+
+
+@router.get("/readings/{reading_id}")
+async def get_reading(reading_id: int):
+    row = db_module.get_tarot_reading(reading_id)
+    if not row:
+        raise HTTPException(status_code=404, detail="reading not found")
+    return row
+
+
+@router.patch("/readings/{reading_id}")
+async def patch_reading(reading_id: int, req: TarotPatchRequest):
+    row = db_module.get_tarot_reading(reading_id)
+    if not row:
+        raise HTTPException(status_code=404, detail="reading not found")
+    db_module.update_tarot_reading(reading_id, **req.model_dump(exclude_none=True))
+    return {"ok": True}
+
+
+@router.delete("/readings/{reading_id}")
+async def delete_reading(reading_id: int):
+    row = db_module.get_tarot_reading(reading_id)
+    if not row:
+        raise HTTPException(status_code=404, detail="reading not found")
+    db_module.delete_tarot_reading(reading_id)
+    return {"ok": True}

agent-office/app/tarot/__init__.py

@@ -0,0 +1 @@
+"""Tarot Lab — Claude Sonnet 기반 evidence·interactions 해석 파이프라인."""

agent-office/app/tarot/pipeline.py

@@ -0,0 +1,139 @@
+"""Tarot 파이프라인 — Claude Sonnet 호출 + 파싱 폴백 + reroll 1회."""
+import json
+import logging
+import time
+from typing import Any, Dict
+
+import httpx
+
+from ..config import (
+    ANTHROPIC_API_KEY,
+    TAROT_MODEL,
+    TAROT_COST_INPUT_PER_M,
+    TAROT_COST_OUTPUT_PER_M,
+    TAROT_TIMEOUT_SEC,
+)
+
+
+logger = logging.getLogger("agent-office.tarot")
+from ..models import TarotInterpretRequest
+from .prompt import SYSTEM_PROMPT, build_user_message
+from .schema import validate_interpretation
+
+
+API_URL = "https://api.anthropic.com/v1/messages"
+
+
+class TarotError(Exception):
+    pass
+
+
+def calc_cost(tokens_in: int, tokens_out: int) -> float:
+    return (
+        tokens_in / 1_000_000 * TAROT_COST_INPUT_PER_M
+        + tokens_out / 1_000_000 * TAROT_COST_OUTPUT_PER_M
+    )
+
+
+def _strip_codeblock(text: str) -> str:
+    t = text.strip()
+    if t.startswith("```"):
+        t = t.strip("`")
+        if t.startswith("json"):
+            t = t[4:]
+        t = t.strip()
+    return t
+
+
+def _extract_json(raw: str) -> dict:
+    cleaned = _strip_codeblock(raw)
+    try:
+        return json.loads(cleaned)
+    except json.JSONDecodeError:
+        start, end = cleaned.find("{"), cleaned.rfind("}")
+        if start >= 0 and end > start:
+            try:
+                return json.loads(cleaned[start : end + 1])
+            except json.JSONDecodeError:
+                pass
+        raise
+
+
+async def _call_claude(user_text: str, feedback: str = "") -> tuple[dict, dict, str]:
+    if not ANTHROPIC_API_KEY:
+        raise TarotError("ANTHROPIC_API_KEY missing")
+    if feedback:
+        user_text = f"이전 응답이 다음 이유로 거절됨: {feedback}\n올바른 스키마(시스템 지침)로 다시 응답.\n\n{user_text}"
+    payload = {
+        "model": TAROT_MODEL,
+        "max_tokens": 1400,  # 응답 시간 단축 — 3-card spread evidence·interactions 포함 충분
+        "system": [{"type": "text", "text": SYSTEM_PROMPT,
+                    "cache_control": {"type": "ephemeral"}}],
+        "messages": [{"role": "user", "content": [{"type": "text", "text": user_text}]}],
+    }
+    headers = {
+        "x-api-key": ANTHROPIC_API_KEY,
+        "anthropic-version": "2023-06-01",
+        "anthropic-beta": "prompt-caching-2024-07-31",
+        "content-type": "application/json",
+    }
+    started = time.monotonic()
+    async with httpx.AsyncClient(timeout=TAROT_TIMEOUT_SEC) as client:
+        r = await client.post(API_URL, headers=headers, json=payload)
+        r.raise_for_status()
+        resp = r.json()
+    latency_ms = int((time.monotonic() - started) * 1000)
+    raw_text = "".join(
+        b.get("text", "") for b in resp.get("content", []) if b.get("type") == "text"
+    )
+    usage = resp.get("usage", {}) or {}
+    tokens_in = int(usage.get("input_tokens", 0) or 0)
+    tokens_out = int(usage.get("output_tokens", 0) or 0)
+    logger.info("tarot claude call: latency=%dms, in=%d, out=%d", latency_ms, tokens_in, tokens_out)
+    parsed = _extract_json(raw_text)
+    meta = {
+        "tokens_in": tokens_in,
+        "tokens_out": tokens_out,
+        "latency_ms": latency_ms,
+    }
+    return parsed, meta, raw_text
+
+
+async def interpret(req: TarotInterpretRequest) -> Dict[str, Any]:
+    user_text = build_user_message(
+        question=req.question or "",
+        category=req.category or "",
+        spread_type=req.spread_type,
+        cards_reference=req.cards_reference,
+        context_meta=req.context_meta or {},
+        spread_count=len(req.cards),
+    )
+
+    total_in, total_out, total_latency = 0, 0, 0
+    last_error = ""
+    for attempt in range(2):
+        try:
+            parsed, meta, _raw = await _call_claude(user_text, feedback=last_error)
+        except httpx.HTTPError as e:
+            raise TarotError(f"Claude HTTP error: {e}") from e
+        except json.JSONDecodeError as e:
+            last_error = f"JSON 파싱 실패: {e}"
+            continue
+        total_in += meta["tokens_in"]
+        total_out += meta["tokens_out"]
+        total_latency += meta["latency_ms"]
+
+        ok, err = validate_interpretation(parsed, req.spread_type)
+        if ok:
+            return {
+                "interpretation_json": parsed,
+                "model": TAROT_MODEL,
+                "tokens_in": total_in,
+                "tokens_out": total_out,
+                "cost_usd": calc_cost(total_in, total_out),
+                "latency_ms": total_latency,
+                "reroll_count": attempt,
+            }
+        last_error = err
+
+    raise TarotError(f"검증 실패 (reroll 2회): {last_error}")

agent-office/app/tarot/prompt.py

@@ -0,0 +1,108 @@
+"""Tarot 프롬프트 — SYSTEM + build_user_message."""
+
+SYSTEM_PROMPT = """당신은 라이더-웨이트(RWS) 타로 덱의 전통 상징체계에 정통한 타로 리더입니다.
+사용자의 질문, 카테고리, 뽑힌 카드 각각의 정·역방향과 위치를 받아 근거 기반으로 해석합니다.
+
+# 해석 원칙
+1. 데이터 우선: "참고 카드 정보" 블록의 키워드·기본의미·상징만을 1차 근거로 사용.
+   외부 변형 의미·다른 덱 해석은 사용하지 않음.
+2. 위치 의미 결합: 카드의 의미와 위치(과거/현재/미래 또는 오늘)를 명시적으로 결합해서 해석. evidence에 근거 기록.
+3. 카드 간 상호작용 분석 (3장 스프레드):
+   - 시너지: 같은 슈트, 같은 원소, 메이저 비율, 정·역 흐름
+   - 충돌·전환: 슈트 충돌(컵-소드, 완드-펜타클), 정→역 전환, 메이저↔마이너 전환
+4. 자기 성찰 톤: 운명론 단정 금지. "…할 가능성이 있어 보입니다" 같은 표현.
+5. 카테고리 컨텍스트: 동일 카드라도 카테고리에 따라 강조점이 달라야 함.
+6. 질문 직접 응답: 사용자 질문을 evidence·advice에서 인용·반영.
+
+# 응답 형식 (strict JSON only — 코드블록 없이 raw JSON)
+{
+  "summary": "전체 흐름 한 단락 (3~4문장)",
+  "cards": [
+    {
+      "position": "<위치 라벨>",
+      "card": "<card_id>",
+      "reversed": <bool>,
+      "interpretation": "3~4문장",
+      "evidence": {
+        "card_meaning_used": "참고 카드 정보에서 인용한 키워드·상징",
+        "position_logic": "왜 이 위치에 이렇게 적용되는지 (1~2문장)",
+        "category_lens": "카테고리 관점에서 부각되는 면 (1문장)"
+      },
+      "advice": "1문장"
+    }
+  ],
+  "interactions": [
+    { "type": "synergy"|"conflict"|"transition",
+      "between": ["<card_id>", "<card_id>"],
+      "explanation": "1~2문장" }
+  ],
+  "advice": "2문장. interactions를 1개 이상 참조할 것.",
+  "warning": "역방향·충돌 경계 (없으면 null)",
+  "confidence": "high"|"medium"|"low"
+}
+
+# confidence 판정 기준
+- high: 3장 모두 한 방향 서사 또는 명확한 전환
+- medium: 2장 일관, 1장 별도 신호
+- low: 카드 간 의미 충돌이 커서 명확한 흐름 잡기 어려움
+
+# 금지사항
+- 참고 카드 정보에 없는 상징 도입 금지
+- 역방향 카드를 정방향처럼 다루지 말 것
+- "신비롭게 들리는" 문구로 채우지 말 것 — evidence에 인용·근거 명시
+- JSON 외 텍스트 금지
+"""
+
+
+SPREAD_NAMES = {
+    "one_card": "오늘의 카드",
+    "three_card": "3장 스프레드 (과거·현재·미래)",
+}
+
+
+def build_user_message(
+    question: str,
+    category: str,
+    spread_type: str,
+    cards_reference: str,
+    context_meta: dict,
+    spread_count: int,
+) -> str:
+    q = question or "(질문 없음)"
+    cat = category or "일반"
+    spread_name = SPREAD_NAMES.get(spread_type, spread_type)
+
+    meta_lines = []
+    if context_meta:
+        if "major_minor_ratio" in context_meta:
+            meta_lines.append(f"- 메이저:마이너 비율: {context_meta['major_minor_ratio']}")
+        if "element_distribution" in context_meta:
+            ed = context_meta["element_distribution"]
+            meta_lines.append(
+                f"- 원소 분포: 공기 {ed.get('air',0)}, 물 {ed.get('water',0)}, 불 {ed.get('fire',0)}, 흙 {ed.get('earth',0)}"
+            )
+        if "orientation_flow" in context_meta:
+            meta_lines.append(f"- 정역 흐름: {context_meta['orientation_flow']}")
+    meta_block = "\n".join(meta_lines) if meta_lines else "(추가 컨텍스트 없음)"
+
+    return f"""# 질문
+{q}
+
+# 카테고리
+{cat}
+
+# 스프레드
+{spread_name} ({spread_count}장)
+
+# 뽑힌 카드와 참고 카드 정보
+{cards_reference}
+
+## 추가 컨텍스트
+{meta_block}
+
+# 작업
+위 정보만을 근거로 사용해, 시스템 지침의 JSON 형식으로 응답하세요.
+- 각 카드의 evidence.card_meaning_used에는 위 "참고 카드 정보"에서 발췌한 키워드·의미를 그대로 인용.
+- interactions는 3장 간 슈트·원소·정역방향 패턴을 분석해 최소 1개 이상 도출 (1장 스프레드면 빈 배열 허용).
+- confidence는 카드 흐름의 일관성에 따라 정직하게 판정.
+"""

agent-office/app/tarot/schema.py

@@ -0,0 +1,36 @@
+"""Tarot 응답 스키마 검증 — 누락·빈 필드 reroll 트리거."""
+
+VALID_CONFIDENCE = {"high", "medium", "low"}
+
+
+def validate_interpretation(parsed: dict, spread_type: str) -> tuple[bool, str]:
+    if not isinstance(parsed, dict):
+        return False, "응답이 dict가 아님"
+    for k in ("summary", "cards", "interactions", "advice", "confidence"):
+        if k not in parsed:
+            return False, f"필수 필드 누락: {k}"
+    if parsed.get("confidence") not in VALID_CONFIDENCE:
+        return False, f"confidence 값 비정상: {parsed.get('confidence')}"
+    cards = parsed.get("cards")
+    if not isinstance(cards, list) or not cards:
+        return False, "cards가 빈 배열"
+    for i, c in enumerate(cards):
+        if not isinstance(c, dict):
+            return False, f"cards[{i}] dict 아님"
+        for k in ("position", "card", "reversed", "interpretation", "advice", "evidence"):
+            if k not in c:
+                return False, f"cards[{i}].{k} 누락"
+        ev = c["evidence"]
+        if not isinstance(ev, dict):
+            return False, f"cards[{i}].evidence dict 아님"
+        for k in ("card_meaning_used", "position_logic", "category_lens"):
+            if k not in ev:
+                return False, f"cards[{i}].evidence.{k} 누락"
+            if not isinstance(ev[k], str) or not ev[k].strip():
+                return False, f"cards[{i}].evidence.{k} 빈 문자열"
+    interactions = parsed.get("interactions")
+    if not isinstance(interactions, list):
+        return False, "interactions가 list 아님"
+    if spread_type == "three_card" and len(interactions) == 0:
+        return False, "three_card는 interactions 1개 이상 필요"
+    return True, ""

agent-office/requirements.txt

@@ -4,5 +4,6 @@ apscheduler==3.10.4
 websockets>=12.0
 httpx>=0.27
 respx>=0.21
+pytest-asyncio>=0.23
 google-api-python-client>=2.100.0
 pytrends>=4.9.2

agent-office/tests/test_insta_keyword_filter.py

@@ -0,0 +1,55 @@
+import os
+import sys
+import tempfile
+
+_fd, _TMP = tempfile.mkstemp(suffix=".db")
+os.close(_fd)
+os.unlink(_TMP)
+os.environ["AGENT_OFFICE_DB_PATH"] = _TMP
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from app.agents.insta import _dedup_and_filter_keywords, KEYWORD_MIN_SCORE
+
+
+def test_filters_below_threshold():
+    """score < 임계값(0.7) 키워드는 제외."""
+    kws = [
+        {"id": 1, "keyword": "금리인하", "category": "경제", "score": 0.9},
+        {"id": 2, "keyword": "환율", "category": "경제", "score": 0.6},  # 컷
+        {"id": 3, "keyword": "반도체", "category": "경제", "score": 0.71},
+    ]
+    out = _dedup_and_filter_keywords(kws, min_score=0.7)
+    kept = {k["keyword"] for k in out}
+    assert kept == {"금리인하", "반도체"}
+
+
+def test_dedup_keeps_highest_score():
+    """동일 keyword 중복 시 최고 score 1개만 유지."""
+    kws = [
+        {"id": 1, "keyword": "AI", "category": "경제", "score": 0.75},
+        {"id": 2, "keyword": "AI", "category": "기술", "score": 0.92},  # 같은 키워드, 더 높음
+    ]
+    out = _dedup_and_filter_keywords(kws, min_score=0.7)
+    assert len(out) == 1
+    assert out[0]["id"] == 2
+    assert out[0]["score"] == 0.92
+
+
+def test_sorted_by_score_desc():
+    kws = [
+        {"id": 1, "keyword": "a", "category": "c", "score": 0.72},
+        {"id": 2, "keyword": "b", "category": "c", "score": 0.95},
+        {"id": 3, "keyword": "c", "category": "c", "score": 0.80},
+    ]
+    out = _dedup_and_filter_keywords(kws, min_score=0.7)
+    assert [k["keyword"] for k in out] == ["b", "c", "a"]
+
+
+def test_empty_when_all_below_threshold():
+    kws = [{"id": 1, "keyword": "x", "category": "c", "score": 0.4}]
+    assert _dedup_and_filter_keywords(kws, min_score=0.7) == []
+
+
+def test_default_threshold_is_0_7():
+    assert KEYWORD_MIN_SCORE == 0.7

agent-office/tests/test_tarot_db.py

@@ -0,0 +1,70 @@
+import json
+import os
+import tempfile
+
+import pytest
+
+from app import db as db_module
+
+
+@pytest.fixture(autouse=True)
+def fresh_db(monkeypatch, tmp_path):
+    db_file = tmp_path / "test_tarot.db"
+    monkeypatch.setattr(db_module, "DB_PATH", str(db_file))
+    db_module.init_db()
+    yield
+    if db_file.exists():
+        db_file.unlink()
+
+
+def test_save_and_get_tarot_reading():
+    rid = db_module.save_tarot_reading({
+        "spread_type": "three_card",
+        "category": "연애",
+        "question": "Q",
+        "cards": [{"position": "과거", "card_id": "the-fool", "reversed": False}],
+        "interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "A", "warning": None, "confidence": "medium"},
+        "model": "claude-sonnet-4-6",
+        "tokens_in": 100, "tokens_out": 200, "cost_usd": 0.005,
+        "confidence": "medium",
+    })
+    assert rid > 0
+    row = db_module.get_tarot_reading(rid)
+    assert row["id"] == rid
+    assert row["category"] == "연애"
+    assert row["interpretation_json"]["summary"] == "S"
+    assert row["favorite"] == 0
+
+
+def test_list_tarot_readings_filters_and_pagination():
+    for cat in ["연애", "연애", "재물"]:
+        db_module.save_tarot_reading({
+            "spread_type": "three_card", "category": cat, "question": "Q",
+            "cards": [], "interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "", "warning": None, "confidence": "low"},
+            "model": "x", "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0, "confidence": "low",
+        })
+    res = db_module.list_tarot_readings(page=1, size=10, category="연애")
+    assert res["total"] == 2
+    assert all(r["category"] == "연애" for r in res["items"])
+
+
+def test_update_tarot_reading_favorite_and_note():
+    rid = db_module.save_tarot_reading({
+        "spread_type": "one_card", "category": None, "question": None,
+        "cards": [], "interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "", "warning": None, "confidence": "high"},
+        "model": "x", "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0, "confidence": "high",
+    })
+    db_module.update_tarot_reading(rid, favorite=True, note="기억하고 싶음")
+    row = db_module.get_tarot_reading(rid)
+    assert row["favorite"] == 1
+    assert row["note"] == "기억하고 싶음"
+
+
+def test_delete_tarot_reading():
+    rid = db_module.save_tarot_reading({
+        "spread_type": "one_card", "category": None, "question": None,
+        "cards": [], "interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "", "warning": None, "confidence": "high"},
+        "model": "x", "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0, "confidence": "high",
+    })
+    db_module.delete_tarot_reading(rid)
+    assert db_module.get_tarot_reading(rid) is None

agent-office/tests/test_tarot_pipeline.py

@@ -0,0 +1,113 @@
+import json
+import pytest
+import respx
+from httpx import Response
+
+from app.tarot import pipeline as p
+from app.models import TarotInterpretRequest
+
+
+def _valid_response_text():
+    return json.dumps({
+        "summary": "S",
+        "cards": [
+            {"position": "과거", "card": "the-fool", "reversed": False,
+             "interpretation": "i", "advice": "a",
+             "evidence": {"card_meaning_used": "k", "position_logic": "p", "category_lens": "c"}},
+            {"position": "현재", "card": "the-lovers", "reversed": True,
+             "interpretation": "i", "advice": "a",
+             "evidence": {"card_meaning_used": "k", "position_logic": "p", "category_lens": "c"}},
+            {"position": "미래", "card": "ten-of-cups", "reversed": False,
+             "interpretation": "i", "advice": "a",
+             "evidence": {"card_meaning_used": "k", "position_logic": "p", "category_lens": "c"}},
+        ],
+        "interactions": [{"type": "synergy", "between": ["the-fool", "ten-of-cups"], "explanation": "."}],
+        "advice": "A", "warning": None, "confidence": "medium",
+    })
+
+
+def _claude_resp(text, in_tok=100, out_tok=200):
+    return {
+        "content": [{"type": "text", "text": text}],
+        "usage": {"input_tokens": in_tok, "output_tokens": out_tok},
+    }
+
+
+def _req():
+    return TarotInterpretRequest(
+        spread_type="three_card",
+        category="연애",
+        question="Q",
+        cards=[
+            {"position": "과거", "card_id": "the-fool", "reversed": False},
+            {"position": "현재", "card_id": "the-lovers", "reversed": True},
+            {"position": "미래", "card_id": "ten-of-cups", "reversed": False},
+        ],
+        cards_reference="REFERENCE",
+        context_meta={"major_minor_ratio": "2:1"},
+    )
+
+
+@pytest.mark.asyncio
+async def test_interpret_happy_path(monkeypatch):
+    monkeypatch.setattr(p, "ANTHROPIC_API_KEY", "sk-test")
+    with respx.mock(base_url="https://api.anthropic.com") as mock:
+        mock.post("/v1/messages").mock(return_value=Response(200, json=_claude_resp(_valid_response_text())))
+        out = await p.interpret(_req())
+    assert out["interpretation_json"]["confidence"] == "medium"
+    assert out["tokens_in"] == 100
+    assert out["tokens_out"] == 200
+    assert out["reroll_count"] == 0
+    assert out["cost_usd"] > 0
+
+
+@pytest.mark.asyncio
+async def test_interpret_codeblock_strip(monkeypatch):
+    monkeypatch.setattr(p, "ANTHROPIC_API_KEY", "sk-test")
+    wrapped = "```json\n" + _valid_response_text() + "\n```"
+    with respx.mock(base_url="https://api.anthropic.com") as mock:
+        mock.post("/v1/messages").mock(return_value=Response(200, json=_claude_resp(wrapped)))
+        out = await p.interpret(_req())
+    assert out["interpretation_json"]["summary"] == "S"
+
+
+@pytest.mark.asyncio
+async def test_interpret_reroll_on_validation_fail(monkeypatch):
+    monkeypatch.setattr(p, "ANTHROPIC_API_KEY", "sk-test")
+    bad = json.loads(_valid_response_text())
+    bad["cards"][0]["evidence"]["card_meaning_used"] = ""
+    bad_text = json.dumps(bad)
+    with respx.mock(base_url="https://api.anthropic.com") as mock:
+        route = mock.post("/v1/messages")
+        route.side_effect = [
+            Response(200, json=_claude_resp(bad_text)),
+            Response(200, json=_claude_resp(_valid_response_text())),
+        ]
+        out = await p.interpret(_req())
+    assert out["reroll_count"] == 1
+    assert out["interpretation_json"]["cards"][0]["evidence"]["card_meaning_used"] == "k"
+
+
+@pytest.mark.asyncio
+async def test_interpret_raises_when_both_attempts_fail(monkeypatch):
+    monkeypatch.setattr(p, "ANTHROPIC_API_KEY", "sk-test")
+    bad = json.loads(_valid_response_text())
+    bad["cards"][0]["evidence"]["card_meaning_used"] = ""
+    bad_text = json.dumps(bad)
+    with respx.mock(base_url="https://api.anthropic.com") as mock:
+        mock.post("/v1/messages").mock(return_value=Response(200, json=_claude_resp(bad_text)))
+        with pytest.raises(p.TarotError):
+            await p.interpret(_req())
+
+
+@pytest.mark.asyncio
+async def test_interpret_raises_when_api_key_missing(monkeypatch):
+    monkeypatch.setattr(p, "ANTHROPIC_API_KEY", "")
+    with pytest.raises(p.TarotError):
+        await p.interpret(_req())
+
+
+def test_calc_cost():
+    assert p.calc_cost(1_000_000, 0) == pytest.approx(3.0)
+    assert p.calc_cost(0, 1_000_000) == pytest.approx(15.0)
+    assert p.calc_cost(500_000, 500_000) == pytest.approx(9.0)

agent-office/tests/test_tarot_routes.py

@@ -0,0 +1,86 @@
+import json
+import pytest
+from fastapi.testclient import TestClient
+
+from app import db as db_module
+
+
+@pytest.fixture(autouse=True)
+def fresh_db(monkeypatch, tmp_path):
+    db_file = tmp_path / "test_routes.db"
+    monkeypatch.setattr(db_module, "DB_PATH", str(db_file))
+    db_module.init_db()
+    from app.main import app
+    yield app
+
+
+def test_interpret_calls_pipeline(monkeypatch, fresh_db):
+    async def fake_interpret(req):
+        return {
+            "interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "A", "warning": None, "confidence": "high"},
+            "model": "claude-sonnet-4-6", "tokens_in": 100, "tokens_out": 200,
+            "cost_usd": 0.005, "latency_ms": 1234, "reroll_count": 0,
+        }
+    from app.tarot import pipeline
+    monkeypatch.setattr(pipeline, "interpret", fake_interpret)
+    client = TestClient(fresh_db)
+    r = client.post("/api/agent-office/tarot/interpret", json={
+        "spread_type": "one_card",
+        "category": "일반",
+        "question": "Q",
+        "cards": [{"position": "오늘", "card_id": "the-fool", "reversed": False}],
+        "cards_reference": "REF",
+        "context_meta": {},
+    })
+    assert r.status_code == 200, r.text
+    assert r.json()["interpretation_json"]["confidence"] == "high"
+
+
+def test_save_and_list(fresh_db):
+    client = TestClient(fresh_db)
+    save = client.post("/api/agent-office/tarot/readings", json={
+        "spread_type": "three_card", "category": "연애", "question": "Q",
+        "cards": [{"position": "과거", "card_id": "the-fool", "reversed": False}],
+        "interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "A", "warning": None, "confidence": "medium"},
+        "model": "claude-sonnet-4-6", "tokens_in": 1, "tokens_out": 2, "cost_usd": 0.01,
+        "confidence": "medium",
+    })
+    assert save.status_code == 200, save.text
+    rid = save.json()["id"]
+    lst = client.get("/api/agent-office/tarot/readings?page=1&size=10")
+    assert lst.json()["total"] == 1
+    assert lst.json()["items"][0]["id"] == rid
+
+
+def test_patch_favorite(fresh_db):
+    client = TestClient(fresh_db)
+    save = client.post("/api/agent-office/tarot/readings", json={
+        "spread_type": "one_card", "cards": [],
+        "interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "A", "warning": None, "confidence": "low"},
+        "model": "x", "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0, "confidence": "low",
+    })
+    rid = save.json()["id"]
+    p = client.patch(f"/api/agent-office/tarot/readings/{rid}", json={"favorite": True})
+    assert p.status_code == 200
+    g = client.get(f"/api/agent-office/tarot/readings/{rid}")
+    assert g.json()["favorite"] == 1
+
+
+def test_delete(fresh_db):
+    client = TestClient(fresh_db)
+    save = client.post("/api/agent-office/tarot/readings", json={
+        "spread_type": "one_card", "cards": [],
+        "interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "A", "warning": None, "confidence": "low"},
+        "model": "x", "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0, "confidence": "low",
+    })
+    rid = save.json()["id"]
+    d = client.delete(f"/api/agent-office/tarot/readings/{rid}")
+    assert d.status_code == 200
+    g = client.get(f"/api/agent-office/tarot/readings/{rid}")
+    assert g.status_code == 404
+
+
+def test_get_missing_reading_404(fresh_db):
+    client = TestClient(fresh_db)
+    r = client.get("/api/agent-office/tarot/readings/99999")
+    assert r.status_code == 404

agent-office/tests/test_tarot_schema.py

@@ -0,0 +1,75 @@
+import pytest
+
+from app.tarot.schema import validate_interpretation
+
+
+def _valid_three():
+    return {
+        "summary": "S",
+        "cards": [
+            {"position": "과거", "card": "the-fool", "reversed": False,
+             "interpretation": "...", "advice": "a",
+             "evidence": {"card_meaning_used": "키워드", "position_logic": "p", "category_lens": "c"}},
+            {"position": "현재", "card": "the-lovers", "reversed": True,
+             "interpretation": "...", "advice": "a",
+             "evidence": {"card_meaning_used": "키워드", "position_logic": "p", "category_lens": "c"}},
+            {"position": "미래", "card": "ten-of-cups", "reversed": False,
+             "interpretation": "...", "advice": "a",
+             "evidence": {"card_meaning_used": "키워드", "position_logic": "p", "category_lens": "c"}},
+        ],
+        "interactions": [{"type": "synergy", "between": ["the-fool", "ten-of-cups"], "explanation": "..."}],
+        "advice": "A",
+        "warning": None,
+        "confidence": "medium",
+    }
+
+
+def test_valid_three_card_passes():
+    ok, msg = validate_interpretation(_valid_three(), "three_card")
+    assert ok, msg
+
+
+def test_missing_evidence_fails():
+    bad = _valid_three()
+    del bad["cards"][0]["evidence"]
+    ok, msg = validate_interpretation(bad, "three_card")
+    assert not ok
+    assert "evidence" in msg
+
+
+def test_empty_card_meaning_used_fails():
+    bad = _valid_three()
+    bad["cards"][0]["evidence"]["card_meaning_used"] = ""
+    ok, msg = validate_interpretation(bad, "three_card")
+    assert not ok
+    assert "card_meaning_used" in msg
+
+
+def test_three_card_requires_interactions():
+    bad = _valid_three()
+    bad["interactions"] = []
+    ok, msg = validate_interpretation(bad, "three_card")
+    assert not ok
+    assert "interactions" in msg
+
+
+def test_one_card_accepts_empty_interactions():
+    one = {
+        "summary": "S",
+        "cards": [{"position": "오늘", "card": "the-fool", "reversed": False,
+                   "interpretation": "...", "advice": "a",
+                   "evidence": {"card_meaning_used": "k", "position_logic": "p", "category_lens": "c"}}],
+        "interactions": [],
+        "advice": "A",
+        "warning": None,
+        "confidence": "high",
+    }
+    ok, msg = validate_interpretation(one, "one_card")
+    assert ok, msg
+
+
+def test_invalid_confidence_fails():
+    bad = _valid_three()
+    bad["confidence"] = "very high"
+    ok, msg = validate_interpretation(bad, "three_card")
+    assert not ok

docker-compose.yml

@@ -113,6 +113,28 @@ services:
       timeout: 5s
       retries: 3
 
+  image-lab:
+    build: ./image-lab
+    container_name: image-lab
+    restart: unless-stopped
+    ports:
+      - "18802:8000"
+    environment:
+      - TZ=${TZ:-Asia/Seoul}
+      - REDIS_URL=${REDIS_URL:-redis://redis:6379}
+      - INTERNAL_API_KEY=${INTERNAL_API_KEY:-}
+      - IMAGE_DATA_DIR=/app/data
+      - CORS_ALLOW_ORIGINS=${CORS_ALLOW_ORIGINS:-http://localhost:3007,http://localhost:8080}
+    volumes:
+      - ${RUNTIME_PATH}/data/image:/app/data
+    depends_on:
+      - redis
+    healthcheck:
+      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"]
+      interval: 60s
+      timeout: 5s
+      retries: 3
+
   insta-lab:
     build:
       context: ./insta-lab
@@ -289,6 +311,7 @@ services:
       - packs-lab
       - travel-proxy
       - video-lab
+      - image-lab
     ports:
       - "8080:80"
     volumes:

docs/superpowers/specs/2026-05-23-tarot-lab-design.md

@@ -0,0 +1,559 @@
+# Tarot Lab v1 — Design Spec
+
+**작성일:** 2026-05-23
+**상태:** 디자인 승인 완료, 구현 계획 작성 대기
+**관련 자산:**
+- `source/images/tarot_page/tarot_main_landing_page.png` (랜딩 시안)
+- `source/images/tarot_page/tarot_card_select_page.png` (카드 선택 시안)
+- `source/images/tarot_page/tarot_background.png` (정적 배경 폴백)
+- `source/images/tarot_page/tarot_cards.png` (카드 콜라주 참고)
+- `source/videos/tarot_main_background.mp4` (히어로 영상)
+
+---
+
+## 1. 목표와 배경
+
+개인 웹 플랫폼에 라이더-웨이트(RWS) 기반 타로 리딩 기능을 추가한다. v1은 **오늘의 카드 / 3장 스프레드 / 리딩 히스토리·마이페이지** 3개 핵심 흐름을 한 번에 배포하고, AI 해석은 Claude Sonnet 4.6을 통해 **근거 기반(evidence)** 으로 생성한다. 켈틱 크로스 10장 스프레드와 카드 78장 정식 이미지 자산은 v2 분리.
+
+### 비목표 (v2 이후)
+- 켈틱 크로스 10장 스프레드
+- 사용자가 제공할 카드 78장 정식 이미지 자산의 정식 매핑 (v1은 placeholder/CSS)
+- 78장 의미 텍스트 완성본 (v1은 메이저 22 + 마이너 키워드만)
+- 텔레그램 자동 push ("매일 오늘의 카드")
+- 카드 78장 도감 화면
+- 즐겨찾기 메모 편집 UI (백엔드 endpoint는 v1에 포함, UI는 v2)
+- **카드 시각 효과 보강** — 카드 이미지 자산 도착 이후 보강:
+  - 카드 hover·focus 시 보더 주변 황금 글로우·sparkle particles
+  - 카드 뒤집기 애니메이션 (3D rotateY transform, 0.6~0.8s ease-out, 뒷면→앞면 전환)
+  - 우주 입자 floating · 별 깜빡임 등 분위기 효과
+  - v1은 hover lift + 단순 fade-in 정도의 미니멀 모션만
+
+---
+
+## 2. 아키텍처
+
+```
+web-ui (React + Vite)
+  /tarot           랜딩 (히어로 영상 + 3-tier)
+  /tarot/today     오늘의 카드 (원카드)
+  /tarot/reading   3장 스프레드 (메인 인터랙션)
+  /tarot/history   마이페이지 (리딩 이력)
+        │
+        │ /api/agent-office/tarot/*
+        ▼
+agent-office (FastAPI 확장)
+  app/routes/tarot.py        4 endpoint
+  app/agents/tarot.py        TarotAgent (Claude Sonnet 호출 + 응답 검증)
+  app/db.py                  tarot_readings 테이블 추가
+        │
+        ▼ Anthropic API
+   Claude Sonnet 4.6
+```
+
+### 경계 결정 이유
+- **카드 78장 메타데이터는 프론트 정적 JSON** — 자주 안 변하고 셔플·선택에 백엔드 호출 불필요. 라운드트립 절약.
+- **AI 해석만 백엔드** — API key 보호 + 호출 로깅·검증·reroll 가능.
+- **히스토리도 백엔드** — localStorage는 기기 의존, 사용자가 영속화 요구.
+- **신규 컨테이너 없음** — agent-office 확장. nginx·docker-compose 변경 0건.
+
+### Why agent-office인가
+1. `ANTHROPIC_API_KEY` 이미 환경변수로 연결됨
+2. Claude SDK + httpx 클라이언트 set up 완료
+3. Agent FSM 패턴(idle→working→reporting)에 자연스럽게 맞음 — TarotAgent도 "리딩 수행" 작업으로 모델링
+4. 텔레그램 봇 연결되어 있어 v2에서 "매일 오늘의 카드" push 확장 여지
+
+---
+
+## 3. 프론트 데이터 모델
+
+### 정적 카드 데이터 (`web-ui/src/pages/tarot/data/cards.js`)
+
+```js
+export const TAROT_DECK = [
+  // Major Arcana 22장
+  {
+    id: 0,
+    slug: "the-fool",
+    name: "바보",
+    nameEn: "The Fool",
+    arcana: "major",
+    element: "air",
+    keywords: ["새로운 시작", "도약", "순수", "자유"],
+    reversedKeywords: ["무모함", "경솔함", "위험", "방향 상실"],
+    meaningUpright: "미지의 세계로 내딛는 첫걸음. 계산보다 직관과 신뢰로 시작하는 시기.",
+    meaningReversed: "준비 없이 뛰어들어 위험을 자초하거나, 두려움으로 첫걸음을 미루는 상태.",
+    image: null,  // 사용자가 /images/tarot/cards/the-fool.png 추가 시 자동 매핑
+  },
+  // ... Major 21장 더
+
+  // Minor Arcana 56장
+  {
+    id: 22,
+    slug: "ace-of-wands",
+    name: "지팡이 에이스",
+    arcana: "minor",
+    suit: "wands",
+    rank: 1,
+    element: "fire",
+    keywords: ["창조의 불씨", "영감", "새로운 시작"],
+    reversedKeywords: ["지연", "동기 부족", "방향 상실"],
+    meaningUpright: "...",
+    meaningReversed: "...",
+    image: null,
+  },
+  // ... Minor 55장 더
+];
+
+export const SPREADS = {
+  one_card: {
+    id: "one_card",
+    name: "오늘의 카드",
+    positions: [{ idx: 0, label: "오늘" }],
+  },
+  three_card: {
+    id: "three_card",
+    name: "3장 스프레드",
+    positions: [
+      { idx: 0, label: "과거" },
+      { idx: 1, label: "현재" },
+      { idx: 2, label: "미래" },
+    ],
+  },
+};
+
+export const CATEGORIES = ["연애", "일·커리어", "관계", "재물", "건강", "일반"];
+```
+
+**v1 시드 데이터 작업량:**
+- 메이저 22장: 정·역 키워드 + 정·역 의미 텍스트 완성 (필수)
+- 마이너 56장: 정·역 키워드만 (필수) + 의미 텍스트는 짧은 요약 1문장씩 (v2에서 보강)
+
+### 카드 이미지 자동 매핑 규칙
+- 사용자가 `web-ui/public/images/tarot/cards/<slug>.png` 추가 시 자동 표시
+- `cards.js`에서 `image: \`/images/tarot/cards/${slug}.png\`` 일관 패턴
+- `onError` → CSS 카드 디자인 폴백 (그라데이션 보더 + 카드명 + 심볼)
+
+---
+
+## 4. 백엔드 데이터 모델
+
+### tarot_readings 테이블 (`agent_office.db`)
+
+```sql
+CREATE TABLE IF NOT EXISTS tarot_readings (
+    id           INTEGER PRIMARY KEY AUTOINCREMENT,
+    created_at   TEXT NOT NULL,            -- UTC ISO8601
+    spread_type  TEXT NOT NULL,            -- 'one_card' | 'three_card'
+    category     TEXT,                     -- '연애' | '일·커리어' | …
+    question     TEXT,                     -- 사용자 입력 (NULL 가능)
+    cards        TEXT NOT NULL,            -- JSON: [{position, card_id, reversed}]
+    interpretation_json TEXT,              -- Claude 응답 파싱 결과 전체
+    summary      TEXT,                     -- interpretation_json.summary 빠른 조회용
+    model        TEXT,                     -- 'claude-sonnet-4-6'
+    tokens_in    INTEGER,
+    tokens_out   INTEGER,
+    cost_usd     REAL,
+    confidence   TEXT,                     -- 'high' | 'medium' | 'low'
+    favorite     INTEGER DEFAULT 0,
+    note         TEXT
+);
+CREATE INDEX idx_tarot_created  ON tarot_readings(created_at DESC);
+CREATE INDEX idx_tarot_favorite ON tarot_readings(favorite, created_at DESC);
+```
+
+**저장 정책:**
+- 모든 리딩은 자동 저장 (사용자가 "저장" 누르지 않아도). 사용자가 별도 액션 없이도 히스토리에서 확인 가능.
+- `favorite` 토글 + `note` 편집은 별도 PATCH 호출
+- 카드는 `card_id`(slug)만 저장 — 실제 이름·의미는 항상 프론트 데이터에서 조회 → 카드 데이터 수정이 과거 이력에 자동 반영
+
+### interpretation_json 구조
+
+```json
+{
+  "summary": "전체 흐름 한 단락 (3~4문장)",
+  "cards": [
+    {
+      "position": "과거",
+      "card": "the-fool",
+      "reversed": false,
+      "interpretation": "이 위치에서 이 카드가 의미하는 바 (3~4문장)",
+      "evidence": {
+        "card_meaning_used": "참고 카드 정보에서 인용한 키워드·상징",
+        "position_logic": "왜 이 의미가 이 위치에 그렇게 적용되는지 (1~2문장)",
+        "category_lens": "카테고리 관점에서 부각되는 면 (1문장)"
+      },
+      "advice": "이 카드가 주는 짧고 구체적인 조언 (1문장)"
+    }
+  ],
+  "interactions": [
+    {
+      "type": "synergy" | "conflict" | "transition",
+      "between": ["the-fool", "the-lovers"],
+      "explanation": "두 카드의 슈트·원소·정역방향 흐름 근거 (1~2문장)"
+    }
+  ],
+  "advice": "3장(또는 1장) 종합 조언 (2문장)",
+  "warning": null,
+  "confidence": "high" | "medium" | "low"
+}
+```
+
+---
+
+## 5. API 명세
+
+### 5.1 `POST /api/agent-office/tarot/interpret`
+AI 해석만 수행 (저장과 분리). 응답 받은 후 사용자가 별도 액션 없으면 자동 저장 호출.
+
+**Request:**
+```json
+{
+  "spread_type": "three_card",
+  "category": "연애",
+  "question": "다음 달 그 사람과의 관계는?",
+  "cards": [
+    { "position": "과거", "card_id": "the-fool", "reversed": false },
+    { "position": "현재", "card_id": "the-lovers", "reversed": true },
+    { "position": "미래", "card_id": "ten-of-cups", "reversed": false }
+  ],
+  "cards_reference": "## 1. 위치: 과거 | 카드: The Fool ...",
+  "context_meta": {
+    "major_minor_ratio": "2:1",
+    "element_distribution": { "air": 2, "water": 1, "fire": 0, "earth": 0 },
+    "orientation_flow": "upright→reversed→upright"
+  }
+}
+```
+
+`cards_reference`와 `context_meta`는 프론트가 `cards.js`를 기반으로 빌드해서 전송. 백엔드가 카드 데이터를 따로 가지고 있을 필요 없음 (DRY).
+
+**Response:** `interpretation_json` 구조 + 호출 메타.
+```json
+{
+  "interpretation_json": { /* 위 4절 구조 */ },
+  "model": "claude-sonnet-4-6",
+  "tokens_in": 712,
+  "tokens_out": 942,
+  "cost_usd": 0.0163,
+  "latency_ms": 5240,
+  "reroll_count": 0
+}
+```
+
+**에러:**
+- 400 — spread_type 미지원 / cards 길이 불일치 / cards_reference 빈 문자열
+- 429 — Anthropic API rate limit
+- 500 — Claude 호출 실패 (Retry-After 헤더 포함) 또는 reroll 2회 모두 실패
+
+### 5.2 `POST /api/agent-office/tarot/readings`
+리딩 저장. interpret 결과를 그대로 + 사용자 컨텍스트.
+
+**Request:**
+```json
+{
+  "spread_type": "three_card",
+  "category": "연애",
+  "question": "...",
+  "cards": [...],
+  "interpretation_json": { ... },
+  "model": "claude-sonnet-4-6",
+  "tokens_in": 712, "tokens_out": 942, "cost_usd": 0.0163,
+  "confidence": "medium"
+}
+```
+
+**Response:** `{ "id": 123, "created_at": "2026-05-23T07:42:11Z" }`
+
+### 5.3 `GET /api/agent-office/tarot/readings`
+페이지네이션 + 필터.
+
+**Query:** `?page=1&size=20&favorite=true&spread_type=three_card&category=연애`
+
+**Response:**
+```json
+{
+  "items": [
+    { "id": 123, "created_at": "...", "spread_type": "three_card",
+      "category": "연애", "question": "...", "cards": [...],
+      "summary": "한 줄 요약", "confidence": "medium", "favorite": 1 }
+  ],
+  "page": 1, "size": 20, "total": 47
+}
+```
+
+### 5.4 `PATCH /api/agent-office/tarot/readings/{id}`
+즐겨찾기 토글·메모.
+
+**Request:** `{ "favorite": true }` 또는 `{ "note": "메모" }`
+
+### 5.5 `DELETE /api/agent-office/tarot/readings/{id}`
+이력 삭제.
+
+### Nginx 라우팅
+변경 없음. 기존 `/api/agent-office/` 매칭에 흡수됨.
+
+---
+
+## 6. AI 프롬프트 설계
+
+### SYSTEM_PROMPT
+
+```text
+당신은 라이더-웨이트(RWS) 타로 덱의 전통 상징체계에 정통한 타로 리더입니다.
+사용자의 질문, 카테고리, 뽑힌 카드 각각의 정·역방향과 위치를 받아 근거 기반으로 해석합니다.
+
+# 해석 원칙
+1. 데이터 우선: "참고 카드 정보" 블록의 키워드·기본의미·상징만을 1차 근거로 사용.
+   외부 변형 의미·다른 덱 해석은 사용하지 않음.
+2. 위치 의미 결합: 카드의 의미와 위치(과거/현재/미래 또는 오늘)를 명시적으로 결합해서 해석. evidence에 근거 기록.
+3. 카드 간 상호작용 분석 (3장 스프레드):
+   - 시너지: 같은 슈트, 같은 원소, 메이저 비율, 정·역 흐름
+   - 충돌·전환: 슈트 충돌(컵-소드, 완드-펜타클), 정→역 전환, 메이저↔마이너 전환
+4. 자기 성찰 톤: 운명론 단정 금지. "…할 가능성이 있어 보입니다" 같은 표현.
+5. 카테고리 컨텍스트: 동일 카드라도 카테고리에 따라 강조점이 달라야 함.
+6. 질문 직접 응답: 사용자 질문을 evidence·advice에서 인용·반영.
+
+# 응답 형식 (strict JSON only — 코드블록 없이 raw JSON)
+{
+  "summary": "전체 흐름 한 단락 (3~4문장)",
+  "cards": [
+    {
+      "position": "<위치 라벨>",
+      "card": "<card_id>",
+      "reversed": <bool>,
+      "interpretation": "3~4문장",
+      "evidence": {
+        "card_meaning_used": "참고 카드 정보에서 인용한 키워드·상징",
+        "position_logic": "왜 이 위치에 이렇게 적용되는지 (1~2문장)",
+        "category_lens": "카테고리 관점에서 부각되는 면 (1문장)"
+      },
+      "advice": "1문장"
+    }
+  ],
+  "interactions": [
+    { "type": "synergy"|"conflict"|"transition",
+      "between": ["<card_id>", "<card_id>"],
+      "explanation": "1~2문장" }
+  ],
+  "advice": "2문장. interactions를 1개 이상 참조할 것.",
+  "warning": "역방향·충돌 경계 (없으면 null)",
+  "confidence": "high"|"medium"|"low"
+}
+
+# confidence 판정 기준
+- high: 3장 모두 한 방향 서사 또는 명확한 전환
+- medium: 2장 일관, 1장 별도 신호
+- low: 카드 간 의미 충돌이 커서 명확한 흐름 잡기 어려움
+
+# 금지사항
+- 참고 카드 정보에 없는 상징 도입 금지
+- 역방향 카드를 정방향처럼 다루지 말 것
+- "신비롭게 들리는" 문구로 채우지 말 것 — evidence에 인용·근거 명시
+- JSON 외 텍스트 금지
+```
+
+### USER_PROMPT_TEMPLATE
+
+```text
+# 질문
+{question}
+
+# 카테고리
+{category}
+
+# 스프레드
+{spread_name} ({spread_count}장)
+
+# 뽑힌 카드와 참고 카드 정보
+{cards_with_reference_block}
+
+# 작업
+위 정보만을 근거로 사용해, 시스템 지침의 JSON 형식으로 응답하세요.
+- 각 카드의 evidence.card_meaning_used에는 위 "참고 카드 정보"에서 발췌한 키워드·의미를 그대로 인용.
+- interactions는 3장 간 슈트·원소·정역방향 패턴을 분석해 최소 1개 이상 도출.
+- confidence는 카드 흐름의 일관성에 따라 정직하게 판정.
+```
+
+### cards_with_reference_block 예시
+
+```
+## 1. 위치: 과거 | 카드: The Fool (정방향)
+- 아르카나: Major (0)
+- 원소: 공기 (Air)
+- 정방향 키워드: 새로운 시작, 도약, 순수, 자유
+- 정방향 의미: 미지의 세계로 내딛는 첫걸음. 계산보다 직관과 신뢰로 시작하는 시기.
+
+## 2. 위치: 현재 | 카드: The Lovers (역방향)
+- 아르카나: Major (6)
+- 원소: 공기 (Air)
+- 역방향 키워드: 관계 갈등, 선택의 어려움
+- 역방향 의미: 두 길 사이에서 머뭇거리거나, 이미 내린 선택의 의구심이 커지는 시기.
+
+## 3. 위치: 미래 | 카드: Ten of Cups (정방향)
+- 아르카나: Minor (Cups, 10)
+- 원소: 물 (Water)
+- 정방향 키워드: 정서적 충만, 가족·공동체의 행복
+- 정방향 의미: 컵 슈트의 완성 단계. 감정적 만족이 안정된 형태로 자리잡는 시기.
+
+## 추가 컨텍스트
+- 메이저:마이너 비율: 2:1 (메이저 우세 → 큰 인생 주제)
+- 원소 분포: 공기 2, 물 1
+- 정역 흐름: 정→역→정 (일시적 정체 후 회복 가능성)
+```
+
+### 응답 검증 (백엔드)
+- `cards[].evidence.card_meaning_used`가 비어있으면 → reroll 1회 (max 1 retry, 총 2회 호출)
+- `interactions`가 비어있고 spread_type == "three_card"이면 → reroll 1회
+- reroll 2회 모두 실패 → 받은 응답 그대로 저장 + log warning + 500 응답
+- JSON 파싱 실패 → codeblock 추출 시도 → raw 추출 시도 → 텍스트 그대로 summary에 박고 cards=[]
+
+### 비용
+- Sonnet 4.6 입력 $3/1M, 출력 $15/1M
+- 회당 입력 ~700, 출력 ~900 토큰
+- 회당 비용 ~$0.015~0.022
+- 환경변수로 가격 오버라이드: `TAROT_COST_INPUT_PER_M`, `TAROT_COST_OUTPUT_PER_M`
+
+---
+
+## 7. UI 흐름
+
+### 7.1 Route 구조
+| Path | 화면 | 컴포넌트 |
+|---|---|---|
+| `/tarot` | 랜딩 | `Tarot.jsx` |
+| `/tarot/today` | 오늘의 카드 | `TodayCard.jsx` |
+| `/tarot/reading` | 3장 스프레드 메인 | `Reading.jsx` |
+| `/tarot/history` | 마이페이지 | `History.jsx` |
+
+### 7.2 랜딩 (`/tarot`)
+- 영상 배경 (`tarot_main_background.mp4` autoplay muted loop, `prefers-reduced-motion` 시 정지 이미지)
+- Overlay: `linear-gradient(rgba(15,4,40,.5) → rgba(15,4,40,.85))`
+- 헤더 sticky nav: 오늘의 카드 / 타로 리딩 / 가이드 / 히스토리
+- Hero: h1 "당신의 오늘을 비추는 타로" + sub + 2 CTA (지금 시작하기 / 오늘의 카드)
+- 3-tier 카드: 🌙 오늘의 운세 / 🃏 3장 스프레드 / ✨ AI 해석 (hover lift)
+
+### 7.3 3장 스프레드 (`/tarot/reading`)
+3-step 진행, 한 화면 안에서 step 전환.
+
+**Step 1 — 질문 입력 (좌측 panel)**
+- 질문 textarea
+- 카테고리 chip 선택 (`CATEGORIES` 중 1개)
+- 스프레드 라디오 (3장 / 1장)
+- [⊃ 카드 셔플하기] 버튼
+
+**Step 2 — 카드 선택 (중앙)**
+- 셔플된 카드 16장 그리드 (4×4, 카드 뒷면)
+- 카드 hover 시 lift + glow
+- 카드 click 시 자리(과거→현재→미래)로 날아가며 flip + 위치 라벨 표시
+- 3장 모두 채워지면 [AI 해석 시작] 버튼 활성
+
+**Step 3 — AI 해석 (우측 panel)**
+- 좌측: 3장 카드 자리 (카드 click으로 우측 panel 전환)
+- 우측 panel: 선택된 카드명 + 키워드 chip + 기본 의미 + AI interpretation + AI evidence(접을 수 있음) + advice
+- 하단: 종합 summary + advice + warning(있을 때) + confidence 배지
+- 액션: [⭐ 즐겨찾기 토글] / [다시 뽑기]
+
+### 7.4 오늘의 카드 (`/tarot/today`)
+- 단일 큰 카드 슬롯 + "운명을 묻다" 버튼
+- 카테고리·질문 옵션 (default = "일반 / 없음")
+- 클릭 → 1장 추출 + flip 애니메이션 + Claude 호출 → 우측 텍스트로 해석 표시
+- 하루 1회 제한은 v1에 없음 (소비 자유)
+
+### 7.5 히스토리 (`/tarot/history`)
+- 카드 리스트형: 날짜 · 스프레드 종류 · 질문 · 카드 미니 · 요약 한 줄 · confidence 배지 · ⭐ 토글
+- 클릭 → 디테일 모달 (원본 해석 전체)
+- 필터: 즐겨찾기만 / 스프레드 종류 / 카테고리
+- 페이지네이션 20개씩
+
+### 7.6 공용 컴포넌트
+- `TarotCard.jsx` — 단일 카드 (앞·뒷면 토글, props: cardId / reversed / size / clickable)
+- `CardGrid.jsx` — 셔플 16장 그리드 (props: deckSlice / onPick)
+- `SpreadSlots.jsx` — 위치별 슬롯 (props: spread / cards)
+- `InterpretationPanel.jsx` — 우측 패널 (카드 의미 + AI 텍스트 + evidence 접기)
+- `useTarotShuffle.js` — Fisher–Yates + 16장 슬라이스 hook
+- `useTarotReading.js` — 카드 선택 상태 + reference 블록 빌더 + AI 호출 + 저장 hook
+
+### 7.7 디자인 토큰
+- 배경 그라데이션: `#0a0420 → #1a0d2e → #2a1648`
+- 금색 액센트: `#d4af37`
+- 카드 보더 글로우: `0 0 24px rgba(212, 175, 55, .35)`
+- 폰트: 본문 기존 / 타이틀 세리프 (Cormorant Garamond + Noto Serif KR 폴백)
+- 네임스페이스: `.tarot-*`
+
+### 7.8 navLinks 추가
+- id: `tarot`, label: `Tarot`, path: `/tarot`, subtitle: `ARCANA`,
+  description: "라이더-웨이트 카드로 오늘과 내일을 비추는 리딩 랩",
+  icon: sparkle 아이콘, accent: `#a78bfa`
+
+---
+
+## 8. 미디어 자산
+
+### 히어로 영상
+- 원본: `source/videos/tarot_main_background.mp4`
+- 배포 위치: `web-ui/public/videos/tarot_hero.mp4` (Vite public/ 직접 서빙)
+- 권장 압축: 1920×1080 H.264 ≤4Mbps, ≤15초 loop
+- 폴백: `prefers-reduced-motion` 또는 `navigator.connection.saveData` 시 `tarot_background.png` 정지 이미지
+
+### 배경 이미지
+- 원본: `source/images/tarot_page/tarot_background.png`
+- 배포 위치: `web-ui/public/images/tarot_background.png`
+- 사용: 영상 fallback + 카드 선택 페이지 배경 layer
+
+### 카드 자산
+- v1: `web-ui/public/images/tarot/card_back.svg` — 단일 카드 뒷면 SVG (보라+금 + ARCANA TAROT 모노그램)
+- v1 카드 앞면: 78장 모두 CSS 카드 디자인 (그라데이션 보더 + 카드명 세리프 + 심볼 이모지)
+- 사용자 자산 추가 시: `web-ui/public/images/tarot/cards/<slug>.png` 자동 매핑, 누락 시 `onError` → CSS 폴백
+- 정적 파일이므로 이미지 추가 후 별도 빌드 불필요. NAS의 `frontend/images/tarot/cards/`에 robocopy 또는 직접 업로드 → 페이지 reload만으로 즉시 반영
+- 사용자가 78장을 한 번에 추가하지 않아도 됨 — 매핑된 것은 이미지로, 안 된 것은 CSS 폴백으로 자연스럽게 혼용
+
+---
+
+## 9. 테스트 전략
+
+### 프론트 (Vitest)
+- `data/cards.js` 검증: 78장 총수, slug 중복 없음, 메이저 22 + 마이너 56, 모든 카드 keywords·meaningUpright·meaningReversed 존재
+- `useTarotShuffle.js`: Fisher–Yates 정확성 (중복 없음, 분포)
+- `useTarotReading.js`: 카드 선택 상태 전환, reference 블록 빌더 단위 테스트
+- `TarotCard.jsx`: 정·역 토글, flip 상태, 이미지 onError 폴백
+- `Reading.jsx`: step 1→2→3 전환
+
+### 백엔드 (pytest)
+- `tarot.py::interpret`: 응답 파싱 (raw JSON / codeblock 감싸진 JSON / 깨진 JSON 폴백)
+- `tarot.py::interpret`: evidence·interactions 누락 시 reroll 1회 → 실패 시 그대로 저장
+- `db.py`: tarot_readings CRUD 정확성, favorite 필터, 페이지네이션
+- Anthropic 호출은 mock — 실제 호출은 통합 테스트 1건만
+
+### 제외
+- AI 응답 품질 자체는 자동 테스트 불가 — manual QA로 검수
+
+---
+
+## 10. 배포
+
+1. **백엔드 (agent-office 수정만)**: `git push` → Gitea Webhook → agent-office 재빌드 + 자동 마이그레이션 (`CREATE TABLE IF NOT EXISTS`)
+2. **프론트**: 로컬 빌드 → `npm run release:nas` → robocopy (영상·이미지 포함)
+3. **docker-compose 변경 없음**
+4. **nginx 변경 없음**
+5. **`scripts/deploy*.sh` 변경 없음** — 컨테이너 리스트 그대로
+
+---
+
+## 11. 위험·완화
+
+| 위험 | 완화 |
+|---|---|
+| Claude 응답 JSON 깨짐 | 파싱 폴백 3단(codeblock→raw→텍스트) + reroll 1회 |
+| 영상 파일 NAS 트래픽↑ | 압축 후 사이즈 체크 — 5MB 초과 시 사용자 노티 |
+| 카드 이미지 미준비로 임팩트↓ | CSS 카드 디자인을 시안 톤(보라+금)에 맞춰 정교화 |
+| AI 비용 폭주 | 회당 ~$0.02, 일 50회 가정 시 월 ~$30 — 개인 사용 OK |
+| 78장 의미 텍스트 작성 부담 | v1 plan에 별도 "데이터 시드 task" 분리, 메이저 22 우선 + 마이너 키워드만 |
+| reference 블록을 프론트가 빌드 → 백엔드 검증 누락 | reference 블록 빈 문자열·길이 단순 검증만 추가 (carot 검증은 v2) |
+
+---
+
+## 12. v1 작업량 추산
+- 백엔드: agent-office 추가 ~300 LOC (`agents/tarot.py` + `routes/tarot.py` + `db.py` 마이그레이션 + 테스트)
+- 프론트: ~1500~2000 LOC (4 페이지 + 5~7 컴포넌트 + 데이터 + CSS)
+- 카드 시드 데이터: 메이저 22장 완성 + 마이너 56장 키워드만 + 짧은 의미 1문장
+- 예상 plan task: 15~18개

image-lab/Dockerfile

@@ -0,0 +1,7 @@
+FROM python:3.12-slim
+WORKDIR /app
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+COPY app ./app
+EXPOSE 8000
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]

image-lab/app/auth.py

@@ -0,0 +1,13 @@
+"""Windows image-render worker → NAS image-lab internal webhook 인증."""
+from __future__ import annotations
+
+import os
+from fastapi import Header, HTTPException
+
+
+def verify_internal_key(x_internal_key: str = Header(...)):
+    expected = os.getenv("INTERNAL_API_KEY")
+    if not expected:
+        raise HTTPException(401, "INTERNAL_API_KEY not configured on server")
+    if x_internal_key != expected:
+        raise HTTPException(401, "Invalid X-Internal-Key")

image-lab/app/db.py

@@ -0,0 +1,83 @@
+"""SQLite persistence for image_tasks. Single table — task 단위 추적만."""
+from __future__ import annotations
+
+import json
+import os
+import sqlite3
+from contextlib import contextmanager
+from typing import Any, Dict, Optional
+
+DB_PATH = os.path.join(os.getenv("IMAGE_DATA_DIR", "/app/data"), "image.db")
+
+
+@contextmanager
+def _conn():
+    os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    conn.execute("PRAGMA journal_mode=WAL")
+    conn.execute("PRAGMA busy_timeout=5000")
+    try:
+        yield conn
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def init_db() -> None:
+    with _conn() as conn:
+        conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS image_tasks (
+                id          TEXT PRIMARY KEY,
+                provider    TEXT NOT NULL,
+                params      TEXT NOT NULL,
+                status      TEXT DEFAULT 'queued',
+                progress    INTEGER DEFAULT 0,
+                message     TEXT DEFAULT '',
+                image_url   TEXT,
+                error       TEXT,
+                created_at  TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
+                updated_at  TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
+            )
+            """
+        )
+
+
+def _row_to_dict(row) -> Dict[str, Any]:
+    return {
+        "id": row["id"], "provider": row["provider"], "params": row["params"],
+        "status": row["status"], "progress": row["progress"], "message": row["message"],
+        "image_url": row["image_url"], "error": row["error"],
+        "created_at": row["created_at"], "updated_at": row["updated_at"],
+    }
+
+
+def create_task(task_id: str, provider: str, params: Dict[str, Any]) -> Dict[str, Any]:
+    with _conn() as conn:
+        conn.execute(
+            "INSERT INTO image_tasks (id, provider, params) VALUES (?, ?, ?)",
+            (task_id, provider, json.dumps(params)),
+        )
+        row = conn.execute("SELECT * FROM image_tasks WHERE id = ?", (task_id,)).fetchone()
+    return _row_to_dict(row)
+
+
+def update_task(task_id: str, status: str, progress: int, message: str = "",
+                image_url: Optional[str] = None, error: Optional[str] = None) -> None:
+    with _conn() as conn:
+        conn.execute(
+            """
+            UPDATE image_tasks
+            SET status = ?, progress = ?, message = ?, image_url = ?, error = ?,
+                updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')
+            WHERE id = ?
+            """,
+            (status, progress, message, image_url, error, task_id),
+        )
+
+
+def get_task(task_id: str) -> Optional[Dict[str, Any]]:
+    with _conn() as conn:
+        row = conn.execute("SELECT * FROM image_tasks WHERE id = ?", (task_id,)).fetchone()
+    return _row_to_dict(row) if row else None

image-lab/app/internal_router.py

@@ -0,0 +1,52 @@
+"""Windows image-render → NAS image-lab internal webhook.
+
+POST /api/internal/image/update
+- X-Internal-Key 인증 필수
+- image_tasks row update (status, progress, message, image_url, error)
+"""
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel, Field
+
+from . import db
+from .auth import verify_internal_key
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+
+
+class UpdatePayload(BaseModel):
+    task_id: str
+    status: str = Field(..., description="processing|succeeded|failed")
+    progress: int = Field(..., ge=0, le=100)
+    message: str = ""
+    image_url: Optional[str] = None
+    error: Optional[str] = None
+
+
+@router.post(
+    "/api/internal/image/update",
+    dependencies=[Depends(verify_internal_key)],
+)
+def image_update(payload: UpdatePayload):
+    task = db.get_task(payload.task_id)
+    if task is None:
+        raise HTTPException(404, f"task not found: {payload.task_id}")
+
+    db.update_task(
+        payload.task_id,
+        payload.status,
+        payload.progress,
+        message=payload.message,
+        image_url=payload.image_url,
+        error=payload.error,
+    )
+    logger.info(
+        "internal/image/update task=%s status=%s progress=%d",
+        payload.task_id, payload.status, payload.progress,
+    )
+    return {"ok": True}

image-lab/app/main.py

@@ -0,0 +1,113 @@
+"""FastAPI entrypoint for image-lab.
+
+POST /api/image/generate — provider + prompt → Redis push → task_id
+GET  /api/image/tasks/{id} — DB 조회
+GET  /api/image/providers — 3 provider 메타
+"""
+from __future__ import annotations
+
+import json
+import logging
+import os
+import uuid
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, Optional
+
+import redis.asyncio as aioredis
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel, Field
+
+from . import db
+from .internal_router import router as internal_router
+
+logger = logging.getLogger(__name__)
+
+CORS_ALLOW_ORIGINS = os.getenv("CORS_ALLOW_ORIGINS", "http://localhost:3007,http://localhost:8080")
+REDIS_URL = os.getenv("REDIS_URL", "redis://redis:6379")
+redis_client = aioredis.from_url(REDIS_URL, decode_responses=False)
+
+SUPPORTED_PROVIDERS = {"gpt_image", "nano_banana", "flux"}
+
+app = FastAPI()
+app.include_router(internal_router)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=[o.strip() for o in CORS_ALLOW_ORIGINS.split(",")],
+    allow_credentials=False,
+    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS", "PATCH"],
+    allow_headers=["Content-Type"],
+)
+
+
+@app.on_event("startup")
+def on_startup():
+    db.init_db()
+
+
+@app.get("/health")
+def health():
+    return {"ok": True, "service": "image-lab"}
+
+
+@app.get("/api/image/providers")
+def list_providers():
+    """3 provider 항상 노출 (key 누락은 worker가 failed 보고)."""
+    return {"providers": [
+        {"id": "gpt_image",   "name": "GPT Image 2.0", "models": ["gpt-image-1"],
+         "sizes": ["1024x1024", "1024x1536", "1536x1024"]},
+        {"id": "nano_banana", "name": "Nano Banana (Gemini)", "models": ["gemini-2.5-flash-image"],
+         "sizes": ["1024x1024"]},
+        {"id": "flux",        "name": "FLUX (local)", "models": ["flux-schnell", "flux-dev"],
+         "sizes": ["1024x1024", "832x1216", "1216x832"]},
+    ]}
+
+
+class GenerateRequest(BaseModel):
+    provider: str = Field(..., description="gpt_image|nano_banana|flux")
+    model: Optional[str] = None
+    prompt: str
+    size: Optional[str] = None
+    negative_prompt: Optional[str] = None
+    # Provider 별 추가 키는 extra 허용
+    extra: Optional[Dict[str, Any]] = None
+
+    class Config:
+        extra = "allow"
+
+
+async def _push_render_job(task_id: str, job_type: str, params: dict) -> None:
+    """Redis queue:image-render에 push."""
+    kst = timezone(timedelta(hours=9))
+    payload = {
+        "task_id": task_id,
+        "kind": "image",
+        "job_type": job_type,
+        "params": params,
+        "submitted_at": datetime.now(kst).isoformat(),
+    }
+    await redis_client.rpush("queue:image-render", json.dumps(payload))
+
+
+@app.post("/api/image/generate")
+async def generate_image(req: GenerateRequest):
+    """이미지 생성 — Redis 큐로 Windows image-render에 위임."""
+    if req.provider not in SUPPORTED_PROVIDERS:
+        raise HTTPException(400, f"지원하지 않는 provider: {req.provider} (supported: {sorted(SUPPORTED_PROVIDERS)})")
+
+    task_id = str(uuid.uuid4())
+    params = req.model_dump(exclude_none=True)
+    db.create_task(task_id, req.provider, params)
+
+    job_type = f"{req.provider}_generation"   # gpt_image_generation, nano_banana_generation, flux_generation
+    await _push_render_job(task_id, job_type, params)
+    return {"task_id": task_id, "provider": req.provider}
+
+
+@app.get("/api/image/tasks/{task_id}")
+def get_task_status(task_id: str):
+    t = db.get_task(task_id)
+    if not t:
+        raise HTTPException(404, "task not found")
+    return t

image-lab/env.example

@@ -0,0 +1,4 @@
+INTERNAL_API_KEY=replace-me
+IMAGE_DATA_DIR=/app/data
+CORS_ALLOW_ORIGINS=http://localhost:3007,http://localhost:8080
+REDIS_URL=redis://redis:6379

image-lab/requirements.txt

@@ -0,0 +1,5 @@
+fastapi==0.115.0
+uvicorn[standard]==0.30.6
+pydantic==2.9.2
+redis==5.0.8
+httpx==0.27.2

image-lab/tests/test_auth.py

@@ -0,0 +1,19 @@
+import pytest
+from fastapi import HTTPException
+from app.auth import verify_internal_key
+
+def test_no_server_key_rejects(monkeypatch):
+    monkeypatch.delenv("INTERNAL_API_KEY", raising=False)
+    with pytest.raises(HTTPException) as e:
+        verify_internal_key("anything")
+    assert e.value.status_code == 401
+
+def test_wrong_key_rejects(monkeypatch):
+    monkeypatch.setenv("INTERNAL_API_KEY", "secret")
+    with pytest.raises(HTTPException) as e:
+        verify_internal_key("wrong")
+    assert e.value.status_code == 401
+
+def test_correct_key_passes(monkeypatch):
+    monkeypatch.setenv("INTERNAL_API_KEY", "secret")
+    assert verify_internal_key("secret") is None

image-lab/tests/test_db.py

@@ -0,0 +1,29 @@
+import os, tempfile, importlib
+
+def _fresh_db(monkeypatch, tmp):
+    monkeypatch.setenv("IMAGE_DATA_DIR", tmp)
+    import app.db as db
+    importlib.reload(db)
+    db.init_db()
+    return db
+
+def test_create_and_get_task(monkeypatch):
+    with tempfile.TemporaryDirectory() as tmp:
+        db = _fresh_db(monkeypatch, tmp)
+        row = db.create_task("t1", "gpt_image", {"prompt": "a cat"})
+        assert row["id"] == "t1"
+        assert row["provider"] == "gpt_image"
+        assert row["status"] == "queued"
+        got = db.get_task("t1")
+        assert got["id"] == "t1"
+        assert db.get_task("nope") is None
+
+def test_update_task_sets_image_url(monkeypatch):
+    with tempfile.TemporaryDirectory() as tmp:
+        db = _fresh_db(monkeypatch, tmp)
+        db.create_task("t2", "nano_banana", {"prompt": "x"})
+        db.update_task("t2", "succeeded", 100, message="done", image_url="/media/image/t2.png")
+        got = db.get_task("t2")
+        assert got["status"] == "succeeded"
+        assert got["image_url"] == "/media/image/t2.png"
+        assert got["progress"] == 100

image-lab/tests/test_internal_router.py

@@ -0,0 +1,38 @@
+import os, tempfile, importlib
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+def _client(monkeypatch, tmp):
+    monkeypatch.setenv("IMAGE_DATA_DIR", tmp)
+    monkeypatch.setenv("INTERNAL_API_KEY", "secret")
+    import app.db as db; importlib.reload(db); db.init_db()
+    import app.internal_router as ir; importlib.reload(ir)
+    app = FastAPI(); app.include_router(ir.router)
+    return TestClient(app), db
+
+def test_update_requires_key(monkeypatch):
+    with tempfile.TemporaryDirectory() as tmp:
+        client, db = _client(monkeypatch, tmp)
+        db.create_task("t1", "gpt_image", {"prompt": "x"})
+        r = client.post("/api/internal/image/update",
+                        json={"task_id": "t1", "status": "succeeded", "progress": 100})
+        assert r.status_code == 422 or r.status_code == 401  # header 누락
+
+def test_update_succeeds_with_key(monkeypatch):
+    with tempfile.TemporaryDirectory() as tmp:
+        client, db = _client(monkeypatch, tmp)
+        db.create_task("t1", "gpt_image", {"prompt": "x"})
+        r = client.post("/api/internal/image/update",
+                        headers={"X-Internal-Key": "secret"},
+                        json={"task_id": "t1", "status": "succeeded", "progress": 100,
+                              "image_url": "/media/image/t1.png"})
+        assert r.status_code == 200
+        assert db.get_task("t1")["image_url"] == "/media/image/t1.png"
+
+def test_update_unknown_task_404(monkeypatch):
+    with tempfile.TemporaryDirectory() as tmp:
+        client, db = _client(monkeypatch, tmp)
+        r = client.post("/api/internal/image/update",
+                        headers={"X-Internal-Key": "secret"},
+                        json={"task_id": "nope", "status": "failed", "progress": 0})
+        assert r.status_code == 404

image-lab/tests/test_main.py

@@ -0,0 +1,43 @@
+import os, tempfile, importlib
+from fastapi.testclient import TestClient
+
+
+def _client(monkeypatch, tmp):
+    monkeypatch.setenv("IMAGE_DATA_DIR", tmp)
+    import app.db as db
+    importlib.reload(db)
+    db.init_db()
+    import app.main as main
+    importlib.reload(main)
+    pushed = []
+
+    async def fake_push(task_id, job_type, params):
+        pushed.append((task_id, job_type, params))
+
+    monkeypatch.setattr(main, "_push_render_job", fake_push)
+    return TestClient(main.app), db, pushed
+
+
+def test_providers_lists_three(monkeypatch):
+    with tempfile.TemporaryDirectory() as tmp:
+        client, _, _ = _client(monkeypatch, tmp)
+        r = client.get("/api/image/providers")
+        ids = {p["id"] for p in r.json()["providers"]}
+        assert ids == {"gpt_image", "nano_banana", "flux"}
+
+
+def test_generate_rejects_unknown_provider(monkeypatch):
+    with tempfile.TemporaryDirectory() as tmp:
+        client, _, _ = _client(monkeypatch, tmp)
+        r = client.post("/api/image/generate", json={"provider": "midjourney", "prompt": "x"})
+        assert r.status_code == 400
+
+
+def test_generate_creates_task_and_pushes(monkeypatch):
+    with tempfile.TemporaryDirectory() as tmp:
+        client, db, pushed = _client(monkeypatch, tmp)
+        r = client.post("/api/image/generate", json={"provider": "gpt_image", "prompt": "a cat"})
+        assert r.status_code == 200
+        task_id = r.json()["task_id"]
+        assert db.get_task(task_id)["status"] == "queued"
+        assert pushed[0][1] == "gpt_image_generation"

insta-lab/app/main.py

@@ -271,12 +271,40 @@ class TemplateBody(BaseModel):
     description: str = ""
 
 
+def _default_prompt_templates() -> dict:
+    """DB에 저장된 override가 없을 때 노출할 코드 기본값.
+    생성 파이프라인이 실제로 폴백하는 값과 동일한 단일 소스를 사용."""
+    return {
+        "slate_writer": {
+            "template": card_writer.DEFAULT_PROMPT,
+            "description": "카드 10페이지 카피 생성 마스터 프롬프트 (Claude Sonnet). "
+                           "{category}/{keyword}/{articles} 치환자 필수.",
+        },
+        "category_seeds": {
+            "template": json.dumps(DEFAULT_CATEGORY_SEEDS, ensure_ascii=False, indent=2),
+            "description": "트렌드 수집·분류용 카테고리별 시드 키워드 (JSON). "
+                           "최상위 키가 분류 라벨로도 쓰임.",
+        },
+    }
+
+
 @app.get("/api/insta/templates/prompts/{name}")
 def get_prompt(name: str):
     pt = db.get_prompt_template(name)
-    if not pt:
+    if pt:
-        raise HTTPException(404)
         return pt
+    # DB override 없음 → 코드 기본값 노출 (편집 UI가 마스터 프롬프트를 보고 수정 가능)
+    defaults = _default_prompt_templates()
+    if name in defaults:
+        d = defaults[name]
+        return {
+            "name": name,
+            "template": d["template"],
+            "description": d["description"],
+            "updated_at": None,
+            "is_default": True,
+        }
+    raise HTTPException(404)
 
 
 @app.put("/api/insta/templates/prompts/{name}")

insta-lab/tests/test_main_prompt_defaults.py

@@ -0,0 +1,63 @@
+import os
+import gc
+import json
+import tempfile
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app import db as db_module
+
+
+@pytest.fixture
+def client(monkeypatch):
+    fd, path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    monkeypatch.setattr(db_module, "DB_PATH", path)
+    db_module.init_db()
+    from app import main
+    monkeypatch.setattr(main, "DB_PATH", path)
+    with TestClient(main.app) as c:
+        yield c
+    gc.collect()
+    for ext in ("", "-wal", "-shm"):
+        try:
+            os.remove(path + ext)
+        except OSError:
+            pass
+
+
+def test_get_slate_writer_returns_default_when_unset(client):
+    """DB에 없으면 코드 기본 마스터 프롬프트를 200으로 반환 (404 아님)."""
+    resp = client.get("/api/insta/templates/prompts/slate_writer")
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["is_default"] is True
+    assert "{keyword}" in body["template"]
+    assert "{category}" in body["template"]
+
+
+def test_get_category_seeds_returns_default_when_unset(client):
+    """category_seeds 기본값은 유효한 JSON (카테고리→시드 배열)."""
+    resp = client.get("/api/insta/templates/prompts/category_seeds")
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["is_default"] is True
+    seeds = json.loads(body["template"])
+    assert "economy" in seeds and isinstance(seeds["economy"], list)
+
+
+def test_get_unknown_prompt_still_404(client):
+    resp = client.get("/api/insta/templates/prompts/does_not_exist")
+    assert resp.status_code == 404
+
+
+def test_saved_template_overrides_default(client):
+    """PUT로 저장하면 이후 GET은 저장본(is_default 없음)을 반환."""
+    client.put("/api/insta/templates/prompts/slate_writer",
+               json={"template": "내 커스텀 프롬프트", "description": "custom"})
+    resp = client.get("/api/insta/templates/prompts/slate_writer")
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["template"] == "내 커스텀 프롬프트"
+    assert not body.get("is_default")

nginx/default.conf

@@ -276,6 +276,26 @@ server {
     proxy_pass http://$video_internal_backend$request_uri;
   }
 
+  # Video Studio — Windows image-render → NAS image-lab internal webhook
+  # Layer 1·2: nginx IP 화이트리스트 (LAN + Tailscale)
+  # Layer 3: X-Internal-Key (FastAPI dependency)
+  location /api/internal/image/ {
+    allow 192.168.45.0/24;     # LAN 화이트리스트
+    allow 100.64.0.0/10;       # Tailscale CGNAT
+    allow 127.0.0.1;           # NAS 내부
+    deny all;
+
+    resolver 127.0.0.11 valid=10s;
+    set $image_internal_backend image-lab:8000;
+
+    proxy_http_version 1.1;
+    proxy_set_header Host $host;
+    proxy_set_header X-Real-IP $remote_addr;
+    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+    proxy_set_header X-Internal-Key $http_x_internal_key;
+    proxy_pass http://$image_internal_backend$request_uri;
+  }
+
   # portfolio API (Stock) — trailing slash 유무 모두 매칭
   location /api/portfolio {
 	proxy_http_version 1.1;
@@ -337,6 +357,8 @@ server {
     proxy_set_header Upgrade $http_upgrade;
     proxy_set_header Connection "upgrade";
     proxy_read_timeout 86400s;
+    proxy_send_timeout 300s;
+    proxy_connect_timeout 60s;
     proxy_pass http://$agent_office_backend$request_uri;
   }
 

scripts/deploy-nas.sh

@@ -2,7 +2,7 @@
 set -euo pipefail
 
 # ── 서비스 목록 (한 곳에서만 관리) ──
-SERVICES="lotto travel-proxy deployer stock music-lab insta-lab realestate-lab agent-office personal packs-lab video-lab nginx scripts"
+SERVICES="lotto travel-proxy deployer stock music-lab insta-lab realestate-lab agent-office personal packs-lab video-lab image-lab nginx scripts"
 
 # 1. 자동 감지: Docker 컨테이너 내부인가?
 if [ -d "/repo" ] && [ -d "/runtime" ]; then

scripts/deploy.sh

@@ -15,15 +15,15 @@ flock -n 200 || { echo "Deploy already running, skipping"; exit 0; }
 
 # ── 서비스 목록 (한 곳에서만 관리) ──
 # docker compose 서비스명 (deployer 제외 — 자기 자신을 재빌드하면 스크립트 중단)
-BUILD_TARGETS="lotto travel-proxy stock music-lab insta-lab realestate-lab agent-office personal packs-lab video-lab frontend"
+BUILD_TARGETS="lotto travel-proxy stock music-lab insta-lab realestate-lab agent-office personal packs-lab video-lab image-lab frontend"
 # 컨테이너 이름 (고아 정리용 — blog-lab은 폐기 대상으로 정리 리스트에 유지)
-CONTAINER_NAMES="lotto stock music-lab insta-lab blog-lab realestate-lab agent-office personal packs-lab travel-proxy video-lab frontend"
+CONTAINER_NAMES="lotto stock music-lab insta-lab blog-lab realestate-lab agent-office personal packs-lab travel-proxy video-lab image-lab frontend"
 # Infra 서비스 (image-based, 영속 데이터 보존을 위해 stop/rm 없이 up만)
 INFRA_SERVICES="redis"
 # 헬스체크 대상
-HEALTH_ENDPOINTS="lotto stock travel-proxy music-lab insta-lab realestate-lab agent-office personal packs-lab video-lab redis"
+HEALTH_ENDPOINTS="lotto stock travel-proxy music-lab insta-lab realestate-lab agent-office personal packs-lab video-lab image-lab redis"
 # data 디렉토리 (packs-lab은 별도 media/packs 사용)
-DATA_DIRS="music stock insta realestate agent-office personal video"
+DATA_DIRS="music stock insta realestate agent-office personal video image"
 
 # 1. 자동 감지: Docker 컨테이너 내부인가?
 if [ -d "/repo" ] && [ -d "/runtime" ]; then

stock/app/screener/ai_news/analyzer.py

@@ -15,9 +15,15 @@ PROMPT_TEMPLATE = """다음은 종목 {name}({ticker})에 대한 최근 뉴스 {
 
 {news_block}
 
-이 뉴스들이 종목에 호재인지 악재인지 평가하세요.
+이 뉴스들이 종목 주가에 호재인지 악재인지 종합 평가하세요.
-score: -10(매우 강한 악재) ~ +10(매우 강한 호재) 사이의 실수. 0은 중립.
+
-reason: 30자 이내 한 줄 근거.
+규칙:
+- score: -10(매우 강한 악재) ~ +10(매우 강한 호재) 사이의 실수. 명확한 방향성이 없으면 0(중립).
+- 뉴스가 호재·악재로 섞여 있으면 주가에 더 우세한 쪽을 기준으로 부호를 정하세요.
+- reason은 반드시 score 부호와 같은 방향의 근거만 쓰세요.
+  · score가 양수(호재)면 호재 근거만, 음수(악재)면 악재 근거만 적습니다.
+  · 호재 평가에 악재 내용을, 악재 평가에 호재 내용을 섞지 마세요.
+- reason: 30자 이내 한 줄.
 
 JSON으로만 응답하세요. 다른 텍스트 금지:
 {{"score": <float>, "reason": "<string>"}}"""

stock/app/screener/ai_news/pipeline.py

@@ -124,8 +124,10 @@ async def refresh_daily(
     if successes:
         _upsert_news_sentiment(conn, asof, successes, source="articles")
 
-    top_pos = sorted(successes, key=lambda r: -r["score_raw"])[:5]
+    # 부호 게이트: 호재(score>0)·악재(score<0)만 분류. score 미만 종목이 5개 미만이어도
-    top_neg = sorted(successes, key=lambda r: r["score_raw"])[:5]
+    # 반대 부호 종목으로 채우지 않음 (양수 종목이 악재란에 섞이는 문제 방지). 중립(0)은 제외.
+    top_pos = sorted([r for r in successes if r["score_raw"] > 0], key=lambda r: -r["score_raw"])[:5]
+    top_neg = sorted([r for r in successes if r["score_raw"] < 0], key=lambda r: r["score_raw"])[:5]
 
     return {
         "asof": asof.isoformat(),

stock/tests/test_ai_news_pipeline.py

@@ -140,6 +140,71 @@ async def test_refresh_daily_no_match_ticker_skipped(conn):
     assert {r["ticker"] for r in rows} == {"005930"}
 
 
+@pytest.mark.asyncio
+async def test_refresh_daily_sign_gate_no_positive_in_neg(conn):
+    """전 종목 양수 점수면 top_neg는 비어야 함 (호재 종목이 악재란에 채워지면 안 됨)."""
+    asof = dt.date(2026, 5, 13)
+    fake_articles_by_ticker = {
+        "005930": [{"title": "h", "summary": "", "press": "", "pub_date": ""}],
+        "000660": [{"title": "h", "summary": "", "press": "", "pub_date": ""}],
+        "373220": [{"title": "h", "summary": "", "press": "", "pub_date": ""}],
+    }
+    fake_stats = {"total_articles": 3, "matched_pairs": 3, "hit_tickers": 3}
+    scores = {"005930": 6.0, "000660": 2.0, "373220": 0.5}  # 모두 양수
+
+    async def fake_score(llm, ticker, news, *, name=None, model="m"):
+        return {
+            "ticker": ticker, "score_raw": scores[ticker], "reason": "r",
+            "news_count": 1, "tokens_input": 1, "tokens_output": 1, "model": model,
+        }
+
+    with patch.object(pipeline, "articles_source") as mas, \
+         patch.object(pipeline, "_analyzer") as ma, \
+         patch.object(pipeline, "_make_llm") as ml:
+        mas.gather_articles_for_tickers = MagicMock(return_value=(fake_articles_by_ticker, fake_stats))
+        ma.score_sentiment = fake_score
+        ml.return_value.__aenter__.return_value = AsyncMock()
+        ml.return_value.__aexit__.return_value = None
+        result = await pipeline.refresh_daily(conn, asof, concurrency=3)
+
+    assert len(result["top_pos"]) == 3
+    assert result["top_neg"] == []  # 양수 종목이 악재란에 들어가면 안 됨
+
+
+@pytest.mark.asyncio
+async def test_refresh_daily_sign_gate_excludes_neutral(conn):
+    """score=0(중립)은 호재·악재 어디에도 포함되지 않음."""
+    asof = dt.date(2026, 5, 13)
+    fake_articles_by_ticker = {
+        "005930": [{"title": "h", "summary": "", "press": "", "pub_date": ""}],
+        "000660": [{"title": "h", "summary": "", "press": "", "pub_date": ""}],
+        "373220": [{"title": "h", "summary": "", "press": "", "pub_date": ""}],
+    }
+    fake_stats = {"total_articles": 3, "matched_pairs": 3, "hit_tickers": 3}
+    scores = {"005930": 3.0, "000660": 0.0, "373220": -3.0}
+
+    async def fake_score(llm, ticker, news, *, name=None, model="m"):
+        return {
+            "ticker": ticker, "score_raw": scores[ticker], "reason": "r",
+            "news_count": 1, "tokens_input": 1, "tokens_output": 1, "model": model,
+        }
+
+    with patch.object(pipeline, "articles_source") as mas, \
+         patch.object(pipeline, "_analyzer") as ma, \
+         patch.object(pipeline, "_make_llm") as ml:
+        mas.gather_articles_for_tickers = MagicMock(return_value=(fake_articles_by_ticker, fake_stats))
+        ma.score_sentiment = fake_score
+        ml.return_value.__aenter__.return_value = AsyncMock()
+        ml.return_value.__aexit__.return_value = None
+        result = await pipeline.refresh_daily(conn, asof, concurrency=3)
+
+    pos_tickers = {r["ticker"] for r in result["top_pos"]}
+    neg_tickers = {r["ticker"] for r in result["top_neg"]}
+    assert pos_tickers == {"005930"}
+    assert neg_tickers == {"373220"}
+    assert "000660" not in pos_tickers and "000660" not in neg_tickers
+
+
 def test_top_market_cap_tickers(conn):
     out = pipeline._top_market_cap_tickers(conn, n=2)
     assert out == ["005930", "000660"]