feat(insta): dedup_window_days config end-to-end wiring (spec 6.4)

- insta-lab ranked_keywords: add dedup_window_days Query param (default 14, ge=1, le=90); pass to db.list_recent_issued_topics
- service_proxy.insta_ranked: add dedup_window_days param (default 14); include in GET params
- InstaAgent.on_schedule: read dedup_window_days from custom_config (default 14); pass to insta_ranked call
- test_ranked_respects_dedup_window: verifies window param gates eligible flag correctly

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

CLAUDE.md

@@ -318,6 +318,8 @@ docker compose up -d
 | POST | `/api/insta/slates/{id}/render` | 카드 렌더 재시도 |
 | GET | `/api/insta/slates/{id}/assets/{page}` | 카드 PNG 다운로드 (1~10) |
 | GET | `/api/insta/slates/{id}/package` | zip 패키지 (10 PNG + caption.txt) |
+| GET | `/api/insta/keywords/ranked` | 4신호 선별 점수 + eligible (자율 발급용) |
+| POST | `/api/insta/slates/{id}/decision` | 승인/반려 (approved→published) |
 | GET | `/api/insta/tasks/{task_id}` | BackgroundTask 상태 폴링 |
 | GET/PUT | `/api/insta/templates/prompts/{name}` | 프롬프트 템플릿 CRUD |
 | POST | `/api/internal/insta/update` | Windows 워커 결과 webhook |

agent-office/app/agents/insta.py

@@ -71,14 +71,32 @@ class InstaAgent(BaseAgent):
         config = get_agent_config(self.agent_id) or {}
         custom = config.get("custom_config", {}) or {}
         auto_select = bool(custom.get("auto_select", False))
+        autonomous = bool(custom.get("autonomous_issue", False))
+        threshold = float(custom.get("select_threshold", 0.6))
+        max_per_day = int(custom.get("max_per_day", 2))
+        dedup_window_days = int(custom.get("dedup_window_days", 14))
 
         task_id = create_task(self.agent_id, "insta_daily", {"auto_select": auto_select},
                               requires_approval=False)
         await self.transition("working", "뉴스 수집·키워드 추출", task_id)
         try:
-            prefs = await service_proxy.insta_get_preferences()
-            add_log(self.agent_id, f"insta preferences: {prefs}", "info", task_id)
+            try:
+                prefs = await service_proxy.insta_get_preferences()
+                add_log(self.agent_id, f"insta preferences: {prefs}", "info", task_id)
+            except Exception as _pref_err:
+                add_log(self.agent_id, f"insta preferences unavailable: {_pref_err}", "warning", task_id)
             await self._run_collect_and_extract()
+            if autonomous:
+                ranked = await service_proxy.insta_ranked(threshold=threshold, limit=20, dedup_window_days=dedup_window_days)
+                eligible = [r for r in ranked if r.get("eligible")][:max_per_day]
+                if not eligible:
+                    await messaging.send_raw("📰 [인스타 큐레이터] 오늘은 발행할 가치 있는 주제가 없습니다.")
+                else:
+                    for pick in eligible:
+                        await self._generate_and_preview(pick)
+                update_task_status(task_id, "succeeded", {"issued": len(eligible)})
+                await self.transition("idle", "자율 발급 후보 프리뷰 완료")
+                return
             kws = await service_proxy.insta_list_keywords(used=False)
             if auto_select:
                 await self._auto_render(kws)
@@ -161,6 +179,27 @@ class InstaAgent(BaseAgent):
         full_caption = f"{caption}\n\n{hashtags}".strip()
         await _send_media_group(media, caption=full_caption)
 
+    async def _generate_and_preview(self, pick: dict) -> None:
+        """eligible 픽 → 슬레이트 생성·렌더 → 커버 프리뷰 + 승인 버튼."""
+        created = await service_proxy.insta_create_slate(
+            keyword=pick["keyword"], category=pick["category"], keyword_id=pick["id"],
+        )
+        st = await self._wait_task(created["task_id"], step="slate", timeout_sec=600)
+        slate_id = st["result_id"]
+        cover = await service_proxy.insta_get_asset_bytes(slate_id, 1)
+        bd = pick.get("breakdown", {})
+        caption = (f"🎴 <b>{pick['keyword']}</b> ({pick['category']})\n"
+                   f"점수 {pick.get('final_score')} · fresh {bd.get('freshness')} "
+                   f"fit {bd.get('account_fit')} claude {bd.get('claude')}\n승인하시겠어요?")
+        kb = {"inline_keyboard": [[
+            {"text": "✅ 승인", "callback_data": f"issue_approve_{slate_id}"},
+            {"text": "❌ 반려", "callback_data": f"issue_reject_{slate_id}"},
+            {"text": "🔄 재생성", "callback_data": f"issue_regen_{slate_id}"},
+        ]]}
+        await messaging.send_photo(cover, caption=caption, reply_markup=kb)
+        create_task(self.agent_id, "insta_issue", {"slate_id": slate_id, "keyword_id": pick["id"]},
+                    requires_approval=True)
+
     async def on_command(self, command: str, params: dict) -> dict:
         if command == "extract":
             await self._run_collect_and_extract()
@@ -188,6 +227,38 @@ class InstaAgent(BaseAgent):
                 return {"ok": False}
             await self._render_and_push(kid)
             return {"ok": True}
+        if action in ("issue_approve", "issue_reject"):
+            sid = int(params.get("slate_id") or 0)
+            if not sid:
+                return {"ok": False}
+            decision = "approved" if action == "issue_approve" else "rejected"
+            await service_proxy.insta_decision(sid, decision)
+            if decision == "approved":
+                slate = await service_proxy.insta_get_slate(sid)
+                media = []
+                for a in slate["assets"][:10]:
+                    data = await service_proxy.insta_get_asset_bytes(sid, a["page_index"])
+                    media.append({"type": "photo", "_bytes": data})
+                cap = f"{slate.get('suggested_caption','')}\n\n{' '.join(slate.get('hashtags', []) or [])}".strip()
+                await _send_media_group(media, caption=cap)
+                await messaging.send_raw(f"✅ 발행 완료 (slate {sid})")
+            else:
+                await messaging.send_raw(f"❌ 반려됨 (slate {sid})")
+            return {"ok": True}
+        if action == "issue_regen":
+            sid = int(params.get("slate_id") or 0)
+            if not sid:
+                return {"ok": False}
+            slate = await service_proxy.insta_get_slate(sid)
+            await service_proxy.insta_decision(sid, "rejected")
+            await self._generate_and_preview({
+                "id": 0,
+                "keyword": slate["keyword"],
+                "category": slate["category"],
+                "final_score": None,
+                "breakdown": {},
+            })
+            return {"ok": True}
         return {"ok": False}
 
     async def on_approval(self, task_id: str, approved: bool, feedback: str = "") -> None:

agent-office/app/service_proxy.py

@@ -228,6 +228,26 @@ async def insta_put_preferences(weights: Dict[str, float]) -> Dict[str, Any]:
     return resp.json()
 
 
+async def insta_ranked(threshold: float = 0.6, limit: int = 20, dedup_window_days: int = 14) -> list:
+    async with httpx.AsyncClient(timeout=120) as client:
+        r = await client.get(
+            f"{INSTA_LAB_URL}/api/insta/keywords/ranked",
+            params={"threshold": threshold, "limit": limit, "dedup_window_days": dedup_window_days},
+        )
+        r.raise_for_status()
+        return r.json()["items"]
+
+
+async def insta_decision(slate_id: int, decision: str) -> dict:
+    async with httpx.AsyncClient(timeout=30) as client:
+        r = await client.post(
+            f"{INSTA_LAB_URL}/api/insta/slates/{slate_id}/decision",
+            json={"decision": decision},
+        )
+        r.raise_for_status()
+        return r.json()
+
+
 # --- realestate-lab ---
 
 async def realestate_collect() -> Dict[str, Any]:

agent-office/app/telegram/messaging.py

@@ -1,8 +1,11 @@
 """고수준 메시지 전송 API."""
+import json
 import uuid
 from typing import Optional
 
-from ..config import TELEGRAM_CHAT_ID
+import httpx
+
+from ..config import TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID
 from ..db import save_telegram_callback
 from .client import _enabled, api_call
 from .formatter import MessageKind, format_agent_message
@@ -81,3 +84,26 @@ async def send_approval_request(
             {"label": "❌ 거절", "action": "reject"},
         ],
     )
+
+
+async def send_photo(
+    photo_bytes: bytes,
+    caption: str = "",
+    reply_markup: Optional[dict] = None,
+    chat_id: Optional[str] = None,
+) -> dict:
+    """PNG/JPEG 바이트를 sendPhoto로 전송. reply_markup으로 인라인 키보드 첨부 가능."""
+    if not TELEGRAM_BOT_TOKEN:
+        return {"ok": False, "reason": "no token"}
+    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendPhoto"
+    data: dict = {
+        "chat_id": chat_id or TELEGRAM_CHAT_ID,
+        "caption": caption[:1024],
+        "parse_mode": "HTML",
+    }
+    if reply_markup:
+        data["reply_markup"] = json.dumps(reply_markup, ensure_ascii=False)
+    files = {"photo": ("cover.png", photo_bytes, "image/png")}
+    async with httpx.AsyncClient(timeout=60) as client:
+        resp = await client.post(url, data=data, files=files)
+    return resp.json()

agent-office/app/telegram/webhook.py

@@ -40,6 +40,9 @@ async def _handle_callback(callback_query: dict) -> Optional[dict]:
     if callback_id.startswith("render_"):
         return await _handle_insta_render(callback_query, callback_id)
 
+    if callback_id.startswith("issue_"):
+        return await _handle_insta_issue(callback_query, callback_id)
+
     cb = get_telegram_callback(callback_id)
     if not cb:
         return None
@@ -132,6 +135,40 @@ async def _handle_insta_render(callback_query: dict, callback_id: str) -> dict:
         return {"ok": False, "error": str(e)}
 
 
+async def _handle_insta_issue(callback_query: dict, callback_id: str) -> dict:
+    """issue_{approve|reject|regen}_{slate_id} 콜백 → InstaAgent.on_callback.
+
+    callback_data 예시: issue_approve_8, issue_reject_8, issue_regen_8
+    InstaAgent.on_callback("issue_approve" | "issue_reject" | "issue_regen", {"slate_id": <int>}) 로 dispatch.
+    """
+    from .messaging import send_raw
+    from ..agents import AGENT_REGISTRY
+
+    await api_call(
+        "answerCallbackQuery",
+        {"callback_query_id": callback_query["id"], "text": "처리 중..."},
+    )
+
+    try:
+        rest = callback_id.removeprefix("issue_")   # 예: "approve_8"
+        verb, sid = rest.rsplit("_", 1)             # ("approve", "8")
+        slate_id = int(sid)
+    except (ValueError, AttributeError):
+        await send_raw("⚠️ 잘못된 issue 콜백 데이터")
+        return {"ok": False, "error": "invalid_callback_data"}
+
+    agent = AGENT_REGISTRY.get("insta")
+    if not agent:
+        await send_raw("⚠️ insta agent 미등록")
+        return {"ok": False, "error": "agent_missing"}
+
+    try:
+        return await agent.on_callback(f"issue_{verb}", {"slate_id": slate_id})
+    except Exception as e:
+        await send_raw(f"⚠️ issue 콜백 처리 실패: {e}")
+        return {"ok": False, "error": str(e)}
+
+
 async def _handle_message(message: dict, agent_dispatcher) -> Optional[dict]:
     """슬래시 명령 메시지 처리."""
     from .router import parse_command, resolve_agent_command, HELP_TEXT

agent-office/tests/test_insta_autonomous.py

@@ -0,0 +1,169 @@
+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__))))
+
+import pytest
+from unittest.mock import AsyncMock
+from app.agents.insta import InstaAgent
+
+
+@pytest.fixture(autouse=True)
+def _init_db():
+    import gc
+    gc.collect()
+    if os.path.exists(_TMP):
+        os.remove(_TMP)
+    from app.db import init_db
+    init_db()
+    yield
+    gc.collect()
+
+
+@pytest.mark.asyncio
+async def test_autonomous_issue_previews_eligible(monkeypatch):
+    agent = InstaAgent()
+    agent.state = "idle"
+    monkeypatch.setattr("app.agents.insta.get_agent_config",
+                        lambda aid: {"custom_config": {"autonomous_issue": True,
+                                                        "select_threshold": 0.5, "max_per_day": 2}})
+    monkeypatch.setattr(agent, "transition", AsyncMock())
+    monkeypatch.setattr(agent, "_run_collect_and_extract", AsyncMock())
+    monkeypatch.setattr("app.agents.insta.service_proxy.insta_ranked", AsyncMock(return_value=[
+        {"id": 1, "keyword": "금리", "category": "economy", "eligible": True, "final_score": 0.8, "breakdown": {}},
+        {"id": 2, "keyword": "x", "category": "economy", "eligible": False, "final_score": 0.1, "breakdown": {}},
+    ]))
+    preview = AsyncMock()
+    monkeypatch.setattr(agent, "_generate_and_preview", preview)
+    monkeypatch.setattr("app.agents.insta.create_task", lambda *a, **k: "t1")
+    monkeypatch.setattr("app.agents.insta.update_task_status", lambda *a, **k: None)
+    monkeypatch.setattr("app.agents.insta.add_log", lambda *a, **k: None)
+    await agent.on_schedule()
+    assert preview.await_count == 1
+    assert preview.await_args.args[0]["id"] == 1
+
+
+@pytest.mark.asyncio
+async def test_callback_approve_publishes_and_delivers(monkeypatch):
+    agent = InstaAgent()
+    monkeypatch.setattr("app.agents.insta.service_proxy.insta_decision",
+                        AsyncMock(return_value={"status": "published"}))
+    monkeypatch.setattr("app.agents.insta.service_proxy.insta_get_slate", AsyncMock(return_value={
+        "assets": [{"page_index": i} for i in range(1, 11)],
+        "suggested_caption": "cap", "hashtags": ["#a"]}))
+    monkeypatch.setattr("app.agents.insta.service_proxy.insta_get_asset_bytes", AsyncMock(return_value=b"png"))
+    monkeypatch.setattr("app.agents.insta._send_media_group", AsyncMock(return_value={"ok": True}))
+    monkeypatch.setattr("app.agents.insta.messaging.send_raw", AsyncMock())
+    res = await agent.on_callback("issue_approve", {"slate_id": 8})
+    assert res["ok"] is True
+
+
+@pytest.mark.asyncio
+async def test_callback_reject_marks_rejected(monkeypatch):
+    agent = InstaAgent()
+    dec = AsyncMock(return_value={"status": "rejected"})
+    monkeypatch.setattr("app.agents.insta.service_proxy.insta_decision", dec)
+    monkeypatch.setattr("app.agents.insta.messaging.send_raw", AsyncMock())
+    res = await agent.on_callback("issue_reject", {"slate_id": 8})
+    assert res["ok"] is True
+    dec.assert_awaited_once_with(8, "rejected")
+
+
+@pytest.mark.asyncio
+async def test_handle_insta_issue_dispatch(monkeypatch):
+    """_handle_insta_issue: issue_approve_8 → on_callback('issue_approve', {slate_id:8})."""
+    import sys
+    # stub api_call so answerCallbackQuery doesn't hit real Telegram
+    import app.telegram.webhook as wh
+    monkeypatch.setattr(wh, "api_call", AsyncMock(return_value={"ok": True}))
+
+    agent = InstaAgent()
+    on_cb = AsyncMock(return_value={"ok": True})
+    monkeypatch.setattr(agent, "on_callback", on_cb)
+
+    from app.agents import AGENT_REGISTRY
+    old = AGENT_REGISTRY.get("insta")
+    AGENT_REGISTRY["insta"] = agent
+    try:
+        result = await wh._handle_insta_issue(
+            {"id": "cq1", "data": "issue_approve_8"},
+            "issue_approve_8",
+        )
+    finally:
+        if old is None:
+            AGENT_REGISTRY.pop("insta", None)
+        else:
+            AGENT_REGISTRY["insta"] = old
+
+    on_cb.assert_awaited_once_with("issue_approve", {"slate_id": 8})
+    assert result["ok"] is True
+
+
+@pytest.mark.asyncio
+async def test_handle_insta_issue_invalid_data(monkeypatch):
+    """_handle_insta_issue: 잘못된 callback_data → ok=False, error=invalid_callback_data."""
+    import app.telegram.webhook as wh
+    monkeypatch.setattr(wh, "api_call", AsyncMock(return_value={"ok": True}))
+    monkeypatch.setattr("app.telegram.messaging.send_raw", AsyncMock())
+
+    result = await wh._handle_insta_issue(
+        {"id": "cq2", "data": "issue_bad"},
+        "issue_bad",
+    )
+    assert result["ok"] is False
+    assert result["error"] == "invalid_callback_data"
+
+
+@pytest.mark.asyncio
+async def test_backward_compat_non_autonomous_uses_legacy_path(monkeypatch):
+    """autonomous_issue=False, auto_select=False → insta_ranked 미호출, _push_keyword_candidates 호출."""
+    agent = InstaAgent()
+    agent.state = "idle"
+    monkeypatch.setattr("app.agents.insta.get_agent_config",
+                        lambda aid: {"custom_config": {"autonomous_issue": False, "auto_select": False}})
+    monkeypatch.setattr(agent, "transition", AsyncMock())
+    monkeypatch.setattr(agent, "_run_collect_and_extract", AsyncMock())
+    # insta_get_preferences는 try/except 안에 있으므로 예외를 던져도 안전하지만 깔끔하게 mock
+    monkeypatch.setattr("app.agents.insta.service_proxy.insta_get_preferences",
+                        AsyncMock(return_value={}))
+    # 비자율 경로에서 insta_ranked는 호출되면 안 된다
+    ranked = AsyncMock()
+    monkeypatch.setattr("app.agents.insta.service_proxy.insta_ranked", ranked)
+    # insta_list_keywords: 비자율 경로에서 반드시 호출
+    monkeypatch.setattr("app.agents.insta.service_proxy.insta_list_keywords",
+                        AsyncMock(return_value=[]))
+    # auto_select=False → _push_keyword_candidates 경로
+    push = AsyncMock()
+    monkeypatch.setattr(agent, "_push_keyword_candidates", push)
+    gen = AsyncMock()
+    monkeypatch.setattr(agent, "_generate_and_preview", gen)
+    monkeypatch.setattr("app.agents.insta.create_task", lambda *a, **k: "t1")
+    monkeypatch.setattr("app.agents.insta.update_task_status", lambda *a, **k: None)
+    monkeypatch.setattr("app.agents.insta.add_log", lambda *a, **k: None)
+    await agent.on_schedule()
+    ranked.assert_not_awaited()      # 자율 경로(insta_ranked) 미진입 확인
+    gen.assert_not_awaited()         # _generate_and_preview 미호출 확인
+    push.assert_awaited_once()       # 기존 candidate-push 경로 진입 확인
+
+
+@pytest.mark.asyncio
+async def test_callback_regen_rejects_old_and_regenerates(monkeypatch):
+    """issue_regen: 기존 슬레이트 rejected 처리 후 같은 키워드로 _generate_and_preview 재호출."""
+    agent = InstaAgent()
+    monkeypatch.setattr("app.agents.insta.service_proxy.insta_get_slate",
+                        AsyncMock(return_value={"keyword": "금리", "category": "economy"}))
+    dec = AsyncMock(return_value={"status": "rejected"})
+    monkeypatch.setattr("app.agents.insta.service_proxy.insta_decision", dec)
+    gen = AsyncMock()
+    monkeypatch.setattr(agent, "_generate_and_preview", gen)
+    res = await agent.on_callback("issue_regen", {"slate_id": 8})
+    assert res["ok"] is True
+    dec.assert_awaited_once_with(8, "rejected")   # 이전 슬레이트 폐기
+    gen.assert_awaited_once()                      # 같은 키워드로 재생성
+    assert gen.await_args.args[0]["keyword"] == "금리"

docs/superpowers/plans/2026-06-11-insta-autonomous-card-issuance.md

@@ -0,0 +1,980 @@
+# insta 자율 카드 발급 (스마트 에이전트 3번) 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:** InstaAgent가 매일 09:30 발행 가치 있는 주제만 자율 선별(4신호)해 카드를 생성·렌더하고, 카드별 텔레그램 승인 게이트로 사람이 최종 결정한 뒤 발급하며, 발행 상태·이력을 추적한다.
+
+**Architecture:** insta-lab이 선별 점수(`selection.py` + `GET /keywords/ranked`)와 발행 상태머신(`card_slates` 컬럼 + `POST /slates/{id}/decision`)을 소유. agent-office `InstaAgent`가 cron 오케스트레이션 + 텔레그램 승인을 담당. 기존 슬레이트 생성·렌더·전달 흐름 재사용.
+
+**Tech Stack:** Python 3.12 / FastAPI / SQLite / anthropic SDK(Haiku) / httpx / pytest. 기존 패턴: `card_writer.py`(Anthropic 클라이언트), `service_proxy.py`(insta httpx 헬퍼), `telegram/webhook.py`(콜백 prefix 디스패치).
+
+**Spec:** `docs/superpowers/specs/2026-06-11-insta-autonomous-card-issuance-design.md`
+
+---
+
+## File Structure
+
+| 파일 | 변경 | 책임 |
+|------|------|------|
+| `insta-lab/app/db.py` | Modify | `card_slates`에 `published_at`/`decision_at` ALTER + `set_slate_decision`/`list_recent_issued_topics` 헬퍼 |
+| `insta-lab/app/selection.py` | Create | 순수 선별 점수(dedup/freshness/account_fit/combine+threshold) |
+| `insta-lab/app/selection_judge.py` | Create | Claude Haiku 일괄 카드가치 판단(외부 IO 격리) |
+| `insta-lab/app/main.py` | Modify | `GET /api/insta/keywords/ranked`, `POST /api/insta/slates/{id}/decision` |
+| `insta-lab/tests/test_selection.py` | Create | selection 순수 단위테스트 |
+| `insta-lab/tests/test_ranked_decision_api.py` | Create | ranked·decision 엔드포인트 테스트 |
+| `agent-office/app/service_proxy.py` | Modify | `insta_ranked`, `insta_decision` 헬퍼 |
+| `agent-office/app/agents/insta.py` | Modify | 자율 `on_schedule` 분기 + 프리뷰 + `issue_*` 콜백 |
+| `agent-office/app/telegram/webhook.py` | Modify | `issue_approve_/issue_reject_/issue_regen_` 디스패치 |
+| `agent-office/tests/test_insta_autonomous.py` | Create | 자율 on_schedule + 콜백 테스트 |
+| `web-backend/CLAUDE.md` + `memory/service_insta.md` | Modify | API 목록 + 메모리 갱신 |
+
+---
+
+## Task 1: insta-lab DB — 발행 상태 컬럼 + 헬퍼
+
+**Files:**
+- Modify: `insta-lab/app/db.py`
+- Test: `insta-lab/tests/test_db_decision.py` (Create)
+
+- [ ] **Step 1: 실패하는 테스트 작성**
+
+`insta-lab/tests/test_db_decision.py`:
+```python
+import os
+import pytest
+from app import db, config
+
+
+@pytest.fixture
+def fresh_db(tmp_path, monkeypatch):
+    monkeypatch.setattr(config, "DB_PATH", str(tmp_path / "insta.db"))
+    monkeypatch.setattr(db, "DB_PATH", str(tmp_path / "insta.db"))
+    db.init_db()
+
+
+def test_set_slate_decision_approved_publishes(fresh_db):
+    sid = db.add_card_slate({"keyword": "금리", "category": "economy"})
+    db.set_slate_decision(sid, "approved")
+    s = db.get_card_slate(sid)
+    assert s["status"] == "published"
+    assert s["published_at"] is not None
+    assert s["decision_at"] is not None
+
+
+def test_set_slate_decision_rejected(fresh_db):
+    sid = db.add_card_slate({"keyword": "환율", "category": "economy"})
+    db.set_slate_decision(sid, "rejected")
+    s = db.get_card_slate(sid)
+    assert s["status"] == "rejected"
+    assert s["decision_at"] is not None
+    assert s["published_at"] is None
+
+
+def test_set_slate_decision_idempotent(fresh_db):
+    sid = db.add_card_slate({"keyword": "주식", "category": "economy"})
+    db.set_slate_decision(sid, "approved")
+    first = db.get_card_slate(sid)["published_at"]
+    db.set_slate_decision(sid, "approved")  # 재호출 no-op
+    assert db.get_card_slate(sid)["published_at"] == first
+
+
+def test_list_recent_issued_topics(fresh_db):
+    a = db.add_card_slate({"keyword": "금리", "category": "economy"})
+    b = db.add_card_slate({"keyword": "우울증", "category": "psychology"})
+    db.set_slate_decision(a, "published") if False else db.set_slate_decision(a, "approved")
+    db.set_slate_decision(b, "rejected")
+    topics = db.list_recent_issued_topics(window_days=14)
+    pairs = {(t["keyword"], t["category"]) for t in topics}
+    assert ("금리", "economy") in pairs
+    assert ("우울증", "psychology") in pairs
+```
+
+- [ ] **Step 2: 테스트 실패 확인**
+
+Run: `cd insta-lab && PYTHONPATH=.. python -m pytest tests/test_db_decision.py -q`
+Expected: FAIL — `db.set_slate_decision` 미존재 + `published_at` 컬럼 없음.
+
+- [ ] **Step 3: `init_db()`에 idempotent ALTER 추가**
+
+`insta-lab/app/db.py`의 `init_db()` 함수 끝(account_preferences seed 직후)에 추가:
+```python
+        # 발행 상태 컬럼 (idempotent ALTER) — 자율 발급 파이프라인
+        cs_cols = [r[1] for r in conn.execute("PRAGMA table_info(card_slates)").fetchall()]
+        if "published_at" not in cs_cols:
+            conn.execute("ALTER TABLE card_slates ADD COLUMN published_at TEXT")
+        if "decision_at" not in cs_cols:
+            conn.execute("ALTER TABLE card_slates ADD COLUMN decision_at TEXT")
+```
+
+- [ ] **Step 4: 헬퍼 함수 추가**
+
+`insta-lab/app/db.py`의 card_slates 섹션(예: `update_slate_status` 아래)에 추가:
+```python
+def set_slate_decision(slate_id: int, decision: str) -> None:
+    """승인/반려 결정 기록. approved→published(+published_at), rejected→rejected.
+    멱등: 이미 published면 published_at 유지."""
+    now = "strftime('%Y-%m-%dT%H:%M:%fZ','now')"
+    with _conn() as conn:
+        if decision == "approved":
+            conn.execute(
+                f"UPDATE card_slates SET status='published', "
+                f"published_at=COALESCE(published_at, {now}), decision_at={now} "
+                f"WHERE id=?",
+                (slate_id,),
+            )
+        elif decision == "rejected":
+            conn.execute(
+                f"UPDATE card_slates SET status='rejected', decision_at={now} WHERE id=?",
+                (slate_id,),
+            )
+        else:
+            raise ValueError(f"invalid decision: {decision}")
+
+
+def list_recent_issued_topics(window_days: int = 14) -> List[Dict[str, Any]]:
+    """최근 window_days 내 published/rejected 슬레이트의 (keyword, category). dedup용."""
+    with _conn() as conn:
+        rows = conn.execute(
+            "SELECT keyword, category FROM card_slates "
+            "WHERE status IN ('published','rejected') "
+            "AND COALESCE(published_at, decision_at) >= datetime('now', ?)",
+            (f"-{int(window_days)} days",),
+        ).fetchall()
+    return [dict(r) for r in rows]
+```
+
+- [ ] **Step 5: 테스트 통과 확인**
+
+Run: `cd insta-lab && PYTHONPATH=.. python -m pytest tests/test_db_decision.py -q`
+Expected: 4 PASS.
+
+- [ ] **Step 6: 커밋**
+```bash
+cd C:/Users/jaeoh/Desktop/workspace/web-backend
+git add insta-lab/app/db.py insta-lab/tests/test_db_decision.py
+git commit -m "feat(insta-lab): 발행 상태 컬럼 + set_slate_decision/list_recent_issued_topics
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 2: insta-lab — `selection.py` 순수 점수
+
+**Files:**
+- Create: `insta-lab/app/selection.py`
+- Test: `insta-lab/tests/test_selection.py`
+
+- [ ] **Step 1: 실패하는 테스트 작성**
+
+`insta-lab/tests/test_selection.py`:
+```python
+from app.selection import score_candidates
+
+NOW = "2026-06-11T00:00:00Z"
+
+
+def _cand(kid, kw, cat, score, suggested_at):
+    return {"id": kid, "keyword": kw, "category": cat, "score": score, "suggested_at": suggested_at}
+
+
+def test_dedup_excludes_recent_issued():
+    cands = [_cand(1, "금리", "economy", 0.9, "2026-06-11T00:00:00Z")]
+    issued = [{"keyword": "금리", "category": "economy"}]
+    out = score_candidates(cands, issued, prefs={}, claude_scores=None, threshold=0.0, now_iso=NOW)
+    assert out[0]["eligible"] is False  # 최근 발행 주제 제외
+
+
+def test_freshness_recent_higher():
+    fresh = _cand(1, "A", "economy", 0.5, "2026-06-11T00:00:00Z")   # 0h
+    stale = _cand(2, "B", "economy", 0.5, "2026-06-04T00:00:00Z")   # 168h
+    out = {c["id"]: c for c in score_candidates([fresh, stale], [], {}, None, threshold=0.0, now_iso=NOW)}
+    assert out[1]["breakdown"]["freshness"] > out[2]["breakdown"]["freshness"]
+
+
+def test_account_fit_uses_weight():
+    cands = [_cand(1, "A", "economy", 0.8, NOW), _cand(2, "B", "psychology", 0.8, NOW)]
+    prefs = {"economy": 2.0, "psychology": 1.0}
+    out = {c["id"]: c for c in score_candidates(cands, [], prefs, None, threshold=0.0, now_iso=NOW)}
+    assert out[1]["breakdown"]["account_fit"] > out[2]["breakdown"]["account_fit"]
+
+
+def test_threshold_gate():
+    cands = [_cand(1, "A", "economy", 0.1, "2026-06-01T00:00:00Z")]  # 낮은 score+오래됨
+    out = score_candidates(cands, [], {}, None, threshold=0.6, now_iso=NOW)
+    assert out[0]["eligible"] is False
+
+
+def test_claude_missing_renormalizes():
+    # claude_scores=None이면 freshness+account_fit만으로 정규화 (claude 항 제외)
+    cands = [_cand(1, "A", "economy", 1.0, NOW)]
+    out = score_candidates(cands, [], {"economy": 1.0}, None, threshold=0.0, now_iso=NOW)
+    assert out[0]["breakdown"]["claude"] is None
+    assert 0.0 <= out[0]["final_score"] <= 1.0
+
+
+def test_claude_included_when_provided():
+    cands = [_cand(1, "A", "economy", 0.5, NOW)]
+    out = score_candidates(cands, [], {"economy": 1.0}, {1: 1.0}, threshold=0.0, now_iso=NOW)
+    assert out[0]["breakdown"]["claude"] == 1.0
+```
+
+- [ ] **Step 2: 테스트 실패 확인**
+
+Run: `cd insta-lab && PYTHONPATH=.. python -m pytest tests/test_selection.py -q`
+Expected: FAIL — `app.selection` 미존재.
+
+- [ ] **Step 3: `selection.py` 작성**
+
+`insta-lab/app/selection.py`:
+```python
+"""발행 가치 자율 선별 — 순수 점수 함수 (외부 IO 없음, 단위테스트 대상).
+
+신호: dedup(게이트), freshness, account_fit, claude(선택).
+final = 가중합(존재하는 신호만 정규화). eligible = dedup통과 and final>=threshold.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+DEFAULT_WEIGHTS = {"freshness": 0.3, "account_fit": 0.3, "claude": 0.4}
+FRESH_WINDOW_HOURS = 168.0  # 7일 → 0
+
+
+def _parse_iso(s: str) -> datetime:
+    return datetime.fromisoformat(s.replace("Z", "+00:00")).astimezone(timezone.utc)
+
+
+def _norm(kw: str) -> str:
+    return (kw or "").strip().lower()
+
+
+def _is_duplicate(keyword: str, category: str, issued: List[Dict[str, Any]]) -> bool:
+    n = _norm(keyword)
+    if not n:
+        return False
+    for it in issued:
+        if it.get("category") != category:
+            continue
+        m = _norm(it.get("keyword", ""))
+        if not m:
+            continue
+        if n == m or n in m or m in n:
+            return True
+    return False
+
+
+def _freshness(suggested_at: str, now: datetime) -> float:
+    try:
+        hours = (now - _parse_iso(suggested_at)).total_seconds() / 3600.0
+    except Exception:
+        return 0.0
+    return max(0.0, min(1.0, 1.0 - hours / FRESH_WINDOW_HOURS))
+
+
+def score_candidates(
+    candidates: List[Dict[str, Any]],
+    issued_topics: List[Dict[str, Any]],
+    prefs: Dict[str, float],
+    claude_scores: Optional[Dict[int, float]] = None,
+    weights: Optional[Dict[str, float]] = None,
+    threshold: float = 0.6,
+    now_iso: Optional[str] = None,
+) -> List[Dict[str, Any]]:
+    w = weights or DEFAULT_WEIGHTS
+    now = _parse_iso(now_iso) if now_iso else datetime.now(timezone.utc)
+    max_w = max(prefs.values()) if prefs else 1.0
+    out: List[Dict[str, Any]] = []
+    for c in candidates:
+        cat = c.get("category", "")
+        dup = _is_duplicate(c.get("keyword", ""), cat, issued_topics)
+        freshness = _freshness(c.get("suggested_at", ""), now)
+        weight = prefs.get(cat, 1.0)
+        account_fit = max(0.0, min(1.0, (weight / max_w) * float(c.get("score", 0.0))))
+        claude = None
+        if claude_scores is not None and c["id"] in claude_scores:
+            claude = max(0.0, min(1.0, float(claude_scores[c["id"]])))
+        # 존재하는 신호만 가중 정규화
+        parts = [("freshness", freshness), ("account_fit", account_fit)]
+        if claude is not None:
+            parts.append(("claude", claude))
+        total_w = sum(w[name] for name, _ in parts)
+        final = sum(w[name] * val for name, val in parts) / total_w if total_w else 0.0
+        eligible = (not dup) and (final >= threshold)
+        out.append({
+            "id": c["id"], "keyword": c.get("keyword"), "category": cat,
+            "final_score": round(final, 4), "eligible": eligible,
+            "breakdown": {"dedup_excluded": dup, "freshness": round(freshness, 4),
+                          "account_fit": round(account_fit, 4), "claude": claude},
+        })
+    out.sort(key=lambda x: (-x["eligible"], -x["final_score"]))
+    return out
+```
+
+- [ ] **Step 4: 테스트 통과 확인**
+
+Run: `cd insta-lab && PYTHONPATH=.. python -m pytest tests/test_selection.py -q`
+Expected: 6 PASS.
+
+- [ ] **Step 5: 커밋**
+```bash
+git add insta-lab/app/selection.py insta-lab/tests/test_selection.py
+git commit -m "feat(insta-lab): selection.py 순수 선별 점수(4신호)
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 3: insta-lab — Claude 카드가치 판단 (`selection_judge.py`)
+
+**Files:**
+- Create: `insta-lab/app/selection_judge.py`
+- Test: `insta-lab/tests/test_selection_judge.py`
+
+- [ ] **Step 1: 실패하는 테스트 작성**
+
+`insta-lab/tests/test_selection_judge.py`:
+```python
+from app import selection_judge
+
+
+def test_parse_judge_response_ok():
+    raw = '[{"keyword_id": 1, "score": 0.8}, {"keyword_id": 2, "score": 0.3}]'
+    assert selection_judge.parse_judge_response(raw) == {1: 0.8, 2: 0.3}
+
+
+def test_parse_judge_response_codefence():
+    raw = '```json\n[{"keyword_id": 5, "score": 0.5}]\n```'
+    assert selection_judge.parse_judge_response(raw) == {5: 0.5}
+
+
+def test_parse_judge_response_garbage_returns_empty():
+    assert selection_judge.parse_judge_response("not json") == {}
+
+
+def test_judge_candidates_no_key_returns_empty(monkeypatch):
+    monkeypatch.setattr(selection_judge, "ANTHROPIC_API_KEY", "")
+    assert selection_judge.judge_candidates([{"id": 1, "keyword": "x", "category": "economy"}]) == {}
+```
+
+- [ ] **Step 2: 테스트 실패 확인**
+
+Run: `cd insta-lab && PYTHONPATH=.. python -m pytest tests/test_selection_judge.py -q`
+Expected: FAIL — 모듈 미존재.
+
+- [ ] **Step 3: `selection_judge.py` 작성**
+
+`insta-lab/app/selection_judge.py`:
+```python
+"""Claude Haiku 일괄 카드가치 판단. 실패/미설정 시 빈 dict (graceful)."""
+from __future__ import annotations
+
+import json
+import logging
+import re
+from typing import Any, Dict, List
+
+from anthropic import Anthropic
+
+from .config import ANTHROPIC_API_KEY, ANTHROPIC_MODEL_HAIKU
+
+logger = logging.getLogger(__name__)
+
+PROMPT = """다음 인스타 카드뉴스 후보 키워드들을 카드로 만들 가치(흥미·시의성·정보성)와
+리스크(민감·논란)를 종합해 0~1 점수로 평가해라. 코드펜스 없이 JSON 배열로만 출력:
+[{{"keyword_id": <id>, "score": <0~1>}}, ...]
+
+후보:
+{items}"""
+
+
+def _strip_codefence(s: str) -> str:
+    s = s.strip()
+    if s.startswith("```"):
+        s = re.sub(r"^```(?:json)?\s*|\s*```$", "", s).strip()
+    return s
+
+
+def parse_judge_response(raw: str) -> Dict[int, float]:
+    try:
+        data = json.loads(_strip_codefence(raw))
+        return {int(d["keyword_id"]): float(d["score"]) for d in data}
+    except Exception:
+        logger.warning("judge 응답 파싱 실패")
+        return {}
+
+
+def judge_candidates(candidates: List[Dict[str, Any]]) -> Dict[int, float]:
+    if not ANTHROPIC_API_KEY or not candidates:
+        return {}
+    items = "\n".join(f'- id={c["id"]}: {c["keyword"]} ({c["category"]})' for c in candidates)
+    try:
+        client = Anthropic(api_key=ANTHROPIC_API_KEY)
+        resp = client.messages.create(
+            model=ANTHROPIC_MODEL_HAIKU, max_tokens=512,
+            messages=[{"role": "user", "content": PROMPT.format(items=items)}],
+        )
+        return parse_judge_response(resp.content[0].text)
+    except Exception:
+        logger.exception("judge_candidates 호출 실패")
+        return {}
+```
+
+- [ ] **Step 4: 테스트 통과 확인**
+
+Run: `cd insta-lab && PYTHONPATH=.. python -m pytest tests/test_selection_judge.py -q`
+Expected: 4 PASS.
+
+- [ ] **Step 5: 커밋**
+```bash
+git add insta-lab/app/selection_judge.py insta-lab/tests/test_selection_judge.py
+git commit -m "feat(insta-lab): Claude Haiku 카드가치 판단(graceful)
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 4: insta-lab — `GET /api/insta/keywords/ranked`
+
+**Files:**
+- Modify: `insta-lab/app/main.py`
+- Test: `insta-lab/tests/test_ranked_decision_api.py` (Create)
+
+- [ ] **Step 1: 실패하는 테스트 작성**
+
+`insta-lab/tests/test_ranked_decision_api.py`:
+```python
+import pytest
+from fastapi.testclient import TestClient
+from app import db, config, selection_judge
+
+
+@pytest.fixture
+def client(tmp_path, monkeypatch):
+    monkeypatch.setattr(config, "DB_PATH", str(tmp_path / "insta.db"))
+    monkeypatch.setattr(db, "DB_PATH", str(tmp_path / "insta.db"))
+    monkeypatch.setattr(selection_judge, "judge_candidates", lambda c: {})  # Claude mock
+    db.init_db()
+    from app.main import app
+    return TestClient(app)
+
+
+def test_ranked_returns_sorted_eligible(client, monkeypatch):
+    db.add_trending_keyword({"keyword": "금리", "category": "economy", "score": 0.9})
+    r = client.get("/api/insta/keywords/ranked?threshold=0.0&limit=10")
+    assert r.status_code == 200
+    items = r.json()["items"]
+    assert len(items) >= 1
+    assert "final_score" in items[0] and "eligible" in items[0]
+
+
+def test_decision_approve_publishes(client):
+    sid = db.add_card_slate({"keyword": "금리", "category": "economy"})
+    r = client.post(f"/api/insta/slates/{sid}/decision", json={"decision": "approved"})
+    assert r.status_code == 200
+    assert db.get_card_slate(sid)["status"] == "published"
+
+
+def test_decision_reject(client):
+    sid = db.add_card_slate({"keyword": "환율", "category": "economy"})
+    r = client.post(f"/api/insta/slates/{sid}/decision", json={"decision": "rejected"})
+    assert r.status_code == 200
+    assert db.get_card_slate(sid)["status"] == "rejected"
+
+
+def test_decision_invalid_400(client):
+    sid = db.add_card_slate({"keyword": "x", "category": "economy"})
+    r = client.post(f"/api/insta/slates/{sid}/decision", json={"decision": "maybe"})
+    assert r.status_code == 400
+```
+
+- [ ] **Step 2: 테스트 실패 확인**
+
+Run: `cd insta-lab && PYTHONPATH=.. python -m pytest tests/test_ranked_decision_api.py -q`
+Expected: FAIL — 라우트 미존재 (404).
+
+- [ ] **Step 3: ranked 라우트 추가**
+
+`insta-lab/app/main.py` import 블록에 추가:
+```python
+from datetime import datetime, timezone
+from . import selection, selection_judge
+```
+
+`list_keywords` 엔드포인트 아래에 추가:
+```python
+@app.get("/api/insta/keywords/ranked")
+def ranked_keywords(limit: int = Query(20, ge=1, le=100), threshold: float = Query(0.6, ge=0.0, le=1.0)):
+    candidates = db.list_trending_keywords(used=False)
+    if not candidates:
+        return {"items": []}
+    issued = db.list_recent_issued_topics(window_days=14)
+    prefs = {p["category"]: p["weight"] for p in db.get_preferences()}
+    claude_scores = selection_judge.judge_candidates(candidates)
+    now_iso = datetime.now(timezone.utc).isoformat()
+    scored = selection.score_candidates(
+        candidates, issued, prefs, claude_scores=claude_scores,
+        threshold=threshold, now_iso=now_iso,
+    )
+    return {"items": scored[:limit]}
+```
+
+- [ ] **Step 4: decision 라우트 추가**
+
+`insta-lab/app/main.py`의 슬레이트 섹션(예: `delete_slate` 위)에 추가:
+```python
+class DecisionBody(BaseModel):
+    decision: str  # "approved" | "rejected"
+
+
+@app.post("/api/insta/slates/{slate_id}/decision")
+def slate_decision(slate_id: int, body: DecisionBody):
+    if not db.get_card_slate(slate_id):
+        raise HTTPException(404, "slate not found")
+    if body.decision not in ("approved", "rejected"):
+        raise HTTPException(400, "decision must be approved|rejected")
+    db.set_slate_decision(slate_id, body.decision)
+    return db.get_card_slate(slate_id)
+```
+
+- [ ] **Step 5: 테스트 통과 확인**
+
+Run: `cd insta-lab && PYTHONPATH=.. python -m pytest tests/test_ranked_decision_api.py -q`
+Expected: 4 PASS.
+
+- [ ] **Step 6: 전체 insta-lab 회귀 + 커밋**
+```bash
+cd insta-lab && PYTHONPATH=.. python -m pytest tests/ -q   # 전부 PASS 확인
+cd C:/Users/jaeoh/Desktop/workspace/web-backend
+git add insta-lab/app/main.py insta-lab/tests/test_ranked_decision_api.py
+git commit -m "feat(insta-lab): GET /keywords/ranked + POST /slates/{id}/decision
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 5: agent-office — service_proxy 헬퍼
+
+**Files:**
+- Modify: `agent-office/app/service_proxy.py`
+
+> 기존 `insta_*` 헬퍼와 동일 패턴(httpx로 `INSTA_LAB_URL` 호출)을 따른다. Task 3 작업 전 `insta_create_slate`(167행) 본문을 열어 base URL·timeout·client 사용 방식을 그대로 모방할 것.
+
+- [ ] **Step 1: 헬퍼 2개 추가**
+
+`agent-office/app/service_proxy.py`의 insta 헬퍼 묶음 끝(예: `insta_put_preferences` 아래)에 추가 — 기존 헬퍼의 `async with httpx.AsyncClient(...)` / base URL 변수명을 동일하게 사용:
+```python
+async def insta_ranked(threshold: float = 0.6, limit: int = 20) -> list:
+    async with httpx.AsyncClient(timeout=120) as client:
+        r = await client.get(
+            f"{INSTA_LAB_URL}/api/insta/keywords/ranked",
+            params={"threshold": threshold, "limit": limit},
+        )
+        r.raise_for_status()
+        return r.json()["items"]
+
+
+async def insta_decision(slate_id: int, decision: str) -> dict:
+    async with httpx.AsyncClient(timeout=30) as client:
+        r = await client.post(
+            f"{INSTA_LAB_URL}/api/insta/slates/{slate_id}/decision",
+            json={"decision": decision},
+        )
+        r.raise_for_status()
+        return r.json()
+```
+
+> 주의: 기존 헬퍼가 `INSTA_LAB_URL`이 아닌 다른 변수명(예: `_INSTA_BASE`)을 쓰면 그 이름으로 맞출 것. timeout(120s)은 ranked의 Claude 호출 대비 여유.
+
+- [ ] **Step 2: import sanity**
+
+Run: `cd agent-office && PYTHONPATH=.. python -c "from app import service_proxy; print('OK')"`
+Expected: OK (httpx 미설치면 pip install httpx 후).
+
+- [ ] **Step 3: 커밋**
+```bash
+git add agent-office/app/service_proxy.py
+git commit -m "feat(agent-office): service_proxy insta_ranked/insta_decision
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 6: agent-office — InstaAgent 자율 발급 경로 + 프리뷰
+
+**Files:**
+- Modify: `agent-office/app/agents/insta.py`
+- Test: `agent-office/tests/test_insta_autonomous.py` (Create)
+
+- [ ] **Step 1: 실패하는 테스트 작성**
+
+`agent-office/tests/test_insta_autonomous.py`:
+```python
+import pytest
+from unittest.mock import AsyncMock, patch
+from app.agents.insta import InstaAgent
+
+
+@pytest.mark.asyncio
+async def test_autonomous_issue_previews_eligible(monkeypatch):
+    agent = InstaAgent()
+    agent.state = "idle"
+    monkeypatch.setattr("app.agents.insta.get_agent_config",
+                        lambda aid: {"custom_config": {"autonomous_issue": True,
+                                                        "select_threshold": 0.5, "max_per_day": 2}})
+    monkeypatch.setattr(agent, "transition", AsyncMock())
+    monkeypatch.setattr(agent, "_run_collect_and_extract", AsyncMock())
+    sp = "app.agents.insta.service_proxy"
+    monkeypatch.setattr(f"{sp}.insta_ranked", AsyncMock(return_value=[
+        {"id": 1, "keyword": "금리", "category": "economy", "eligible": True, "final_score": 0.8,
+         "breakdown": {}},
+        {"id": 2, "keyword": "x", "category": "economy", "eligible": False, "final_score": 0.1,
+         "breakdown": {}},
+    ]))
+    preview = AsyncMock()
+    monkeypatch.setattr(agent, "_generate_and_preview", preview)
+    monkeypatch.setattr("app.agents.insta.create_task", lambda *a, **k: "t1")
+    monkeypatch.setattr("app.agents.insta.update_task_status", lambda *a, **k: None)
+    monkeypatch.setattr("app.agents.insta.add_log", lambda *a, **k: None)
+
+    await agent.on_schedule()
+    # eligible 1건만 프리뷰
+    assert preview.await_count == 1
+    assert preview.await_args.args[0]["id"] == 1
+```
+
+- [ ] **Step 2: 테스트 실패 확인**
+
+Run: `cd agent-office && PYTHONPATH=.. python -m pytest tests/test_insta_autonomous.py -q`
+Expected: FAIL — 자율 분기/`_generate_and_preview` 미존재.
+
+- [ ] **Step 3: `on_schedule`에 자율 분기 추가**
+
+`agent-office/app/agents/insta.py`의 `on_schedule`에서 `auto_select` 분기 직전에 자율 경로를 추가. `custom` 읽은 직후:
+```python
+        autonomous = bool(custom.get("autonomous_issue", False))
+        threshold = float(custom.get("select_threshold", 0.6))
+        max_per_day = int(custom.get("max_per_day", 2))
+```
+그리고 `add_log(...) → _run_collect_and_extract()` 다음의 분기를 교체:
+```python
+            await self._run_collect_and_extract()
+            if autonomous:
+                ranked = await service_proxy.insta_ranked(threshold=threshold, limit=20)
+                eligible = [r for r in ranked if r.get("eligible")][:max_per_day]
+                if not eligible:
+                    await messaging.send_raw("📰 [인스타 큐레이터] 오늘은 발행할 가치 있는 주제가 없습니다.")
+                else:
+                    for pick in eligible:
+                        await self._generate_and_preview(pick)
+                update_task_status(task_id, "succeeded", {"issued": len(eligible)})
+                await self.transition("idle", "자율 발급 후보 프리뷰 완료")
+                return
+            kws = await service_proxy.insta_list_keywords(used=False)
+            if auto_select:
+                ...  # 기존 유지
+```
+(기존 `kws = ... / if auto_select` 블록은 그대로 둔다.)
+
+- [ ] **Step 4: `_generate_and_preview` 메서드 추가**
+
+`insta.py`에 추가 — 슬레이트 생성·렌더(기존 흐름) 후 커버 프리뷰 발송:
+```python
+    async def _generate_and_preview(self, pick: dict) -> None:
+        """eligible 픽 → 슬레이트 생성·렌더 → 커버 프리뷰 + 승인 버튼."""
+        created = await service_proxy.insta_create_slate(
+            keyword=pick["keyword"], category=pick["category"], keyword_id=pick["id"],
+        )
+        st = await self._wait_task(created["task_id"], step="slate", timeout_sec=600)
+        slate_id = st["result_id"]
+        cover = await service_proxy.insta_get_asset_bytes(slate_id, 1)
+        bd = pick.get("breakdown", {})
+        caption = (f"🎴 <b>{pick['keyword']}</b> ({pick['category']})\n"
+                   f"점수 {pick.get('final_score')} · fresh {bd.get('freshness')} "
+                   f"fit {bd.get('account_fit')} claude {bd.get('claude')}\n승인하시겠어요?")
+        kb = {"inline_keyboard": [[
+            {"text": "✅ 승인", "callback_data": f"issue_approve_{slate_id}"},
+            {"text": "❌ 반려", "callback_data": f"issue_reject_{slate_id}"},
+            {"text": "🔄 재생성", "callback_data": f"issue_regen_{slate_id}"},
+        ]]}
+        await messaging.send_photo(cover, caption=caption, reply_markup=kb)
+        create_task(self.agent_id, "insta_issue", {"slate_id": slate_id, "keyword_id": pick["id"]},
+                    requires_approval=True)
+```
+
+> `messaging.send_photo(bytes, caption, reply_markup)`가 없으면 Task 6.5로 추가(아래). 있으면 그대로 사용.
+
+- [ ] **Step 5: 테스트 통과 확인**
+
+Run: `cd agent-office && PYTHONPATH=.. python -m pytest tests/test_insta_autonomous.py::test_autonomous_issue_previews_eligible -q`
+Expected: PASS.
+
+- [ ] **Step 6: 커밋**
+```bash
+git add agent-office/app/agents/insta.py agent-office/tests/test_insta_autonomous.py
+git commit -m "feat(agent-office): InstaAgent 자율 발급 경로 + 커버 프리뷰
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 6.5: agent-office — `messaging.send_photo` (없을 경우만)
+
+**Files:**
+- Modify: `agent-office/app/telegram/messaging.py`
+
+- [ ] **Step 1: 존재 확인**
+
+Run: `grep -n "def send_photo" agent-office/app/telegram/messaging.py`
+이미 있으면 이 Task 건너뜀.
+
+- [ ] **Step 2: 없으면 추가**
+
+`messaging.py`에 `send_raw` 패턴(TELEGRAM_BOT_TOKEN/CHAT_ID 사용)을 따라 추가:
+```python
+async def send_photo(photo_bytes: bytes, caption: str = "", reply_markup: dict = None) -> dict:
+    if not TELEGRAM_BOT_TOKEN:
+        return {"ok": False, "reason": "no token"}
+    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendPhoto"
+    data = {"chat_id": TELEGRAM_CHAT_ID, "caption": caption[:1024], "parse_mode": "HTML"}
+    if reply_markup:
+        data["reply_markup"] = json.dumps(reply_markup, ensure_ascii=False)
+    files = {"photo": ("cover.png", photo_bytes, "image/png")}
+    async with httpx.AsyncClient(timeout=60) as client:
+        resp = await client.post(url, data=data, files=files)
+    return resp.json()
+```
+(상단에 `import json`, `import httpx`, `from ..config import TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID` 필요분 확인.)
+
+- [ ] **Step 3: 커밋**
+```bash
+git add agent-office/app/telegram/messaging.py
+git commit -m "feat(agent-office): messaging.send_photo (인라인 키보드 첨부 사진)
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 7: agent-office — `issue_*` 콜백 처리
+
+**Files:**
+- Modify: `agent-office/app/agents/insta.py` (`on_callback`)
+- Test: `agent-office/tests/test_insta_autonomous.py` (추가)
+
+- [ ] **Step 1: 실패하는 테스트 추가**
+
+`test_insta_autonomous.py`에 추가:
+```python
+@pytest.mark.asyncio
+async def test_callback_approve_publishes_and_delivers(monkeypatch):
+    agent = InstaAgent()
+    sp = "app.agents.insta.service_proxy"
+    dec = AsyncMock(return_value={"status": "published"})
+    monkeypatch.setattr(f"{sp}.insta_decision", dec)
+    monkeypatch.setattr(f"{sp}.insta_get_slate", AsyncMock(return_value={
+        "assets": [{"page_index": i} for i in range(1, 11)],
+        "suggested_caption": "cap", "hashtags": ["#a"]}))
+    monkeypatch.setattr(f"{sp}.insta_get_asset_bytes", AsyncMock(return_value=b"png"))
+    monkeypatch.setattr("app.agents.insta._send_media_group", AsyncMock(return_value={"ok": True}))
+    monkeypatch.setattr("app.agents.insta.messaging.send_raw", AsyncMock())
+    res = await agent.on_callback("issue_approve", {"slate_id": 8})
+    assert res["ok"] is True
+    dec.assert_awaited_once_with(8, "approved")
+
+
+@pytest.mark.asyncio
+async def test_callback_reject_marks_rejected(monkeypatch):
+    agent = InstaAgent()
+    dec = AsyncMock(return_value={"status": "rejected"})
+    monkeypatch.setattr("app.agents.insta.service_proxy.insta_decision", dec)
+    monkeypatch.setattr("app.agents.insta.messaging.send_raw", AsyncMock())
+    res = await agent.on_callback("issue_reject", {"slate_id": 8})
+    assert res["ok"] is True
+    dec.assert_awaited_once_with(8, "rejected")
+```
+
+- [ ] **Step 2: 테스트 실패 확인**
+
+Run: `cd agent-office && PYTHONPATH=.. python -m pytest tests/test_insta_autonomous.py -q`
+Expected: FAIL — issue_* 액션 미처리.
+
+- [ ] **Step 3: `on_callback`에 issue_* 분기 추가**
+
+`insta.py`의 `on_callback`을 확장:
+```python
+    async def on_callback(self, action: str, params: dict) -> dict:
+        if action == "render":
+            kid = int(params.get("keyword_id") or 0)
+            if not kid:
+                return {"ok": False}
+            await self._render_and_push(kid)
+            return {"ok": True}
+        if action in ("issue_approve", "issue_reject"):
+            sid = int(params.get("slate_id") or 0)
+            if not sid:
+                return {"ok": False}
+            decision = "approved" if action == "issue_approve" else "rejected"
+            await service_proxy.insta_decision(sid, decision)
+            if decision == "approved":
+                slate = await service_proxy.insta_get_slate(sid)
+                media = []
+                for a in slate["assets"][:10]:
+                    data = await service_proxy.insta_get_asset_bytes(sid, a["page_index"])
+                    media.append({"type": "photo", "_bytes": data})
+                cap = f"{slate.get('suggested_caption','')}\n\n{' '.join(slate.get('hashtags',[]) or [])}".strip()
+                await _send_media_group(media, caption=cap)
+                await messaging.send_raw(f"✅ 발행 완료 (slate {sid})")
+            else:
+                await messaging.send_raw(f"❌ 반려됨 (slate {sid})")
+            return {"ok": True}
+        if action == "issue_regen":
+            sid = int(params.get("slate_id") or 0)
+            if not sid:
+                return {"ok": False}
+            slate = await service_proxy.insta_get_slate(sid)
+            await service_proxy.insta_decision(sid, "rejected")  # 이전 폐기
+            await self._generate_and_preview({
+                "id": 0, "keyword": slate["keyword"], "category": slate["category"],
+                "final_score": None, "breakdown": {},
+            })
+            return {"ok": True}
+        return {"ok": False}
+```
+
+> `insta_create_slate`는 `keyword_id` 없이도 동작(기존 시그니처 `keyword_id: Optional`). regen은 keyword_id=0 → mark_keyword_used 생략.
+
+- [ ] **Step 4: 테스트 통과 확인**
+
+Run: `cd agent-office && PYTHONPATH=.. python -m pytest tests/test_insta_autonomous.py -q`
+Expected: 모두 PASS.
+
+- [ ] **Step 5: 커밋**
+```bash
+git add agent-office/app/agents/insta.py agent-office/tests/test_insta_autonomous.py
+git commit -m "feat(agent-office): issue_approve/reject/regen 콜백 처리
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 8: agent-office — 텔레그램 콜백 디스패치
+
+**Files:**
+- Modify: `agent-office/app/telegram/webhook.py`
+
+- [ ] **Step 1: `_handle_callback`에 issue_* 분기 추가**
+
+`webhook.py`의 `_handle_callback`에서 `render_` 분기 아래에 추가:
+```python
+    if callback_id.startswith("issue_"):
+        return await _handle_insta_issue(callback_query, callback_id)
+```
+
+- [ ] **Step 2: `_handle_insta_issue` 추가**
+
+`_handle_insta_render`(103행~)를 본떠 추가:
+```python
+async def _handle_insta_issue(callback_query: dict, callback_id: str) -> dict:
+    """issue_{approve|reject|regen}_{slate_id} → InstaAgent.on_callback."""
+    from ..agents import AGENT_REGISTRY  # _handle_insta_render와 동일 방식으로 에이전트 해석
+    try:
+        rest = callback_id.removeprefix("issue_")          # "approve_8"
+        verb, sid = rest.rsplit("_", 1)
+        slate_id = int(sid)
+    except (ValueError, AttributeError):
+        return {"ok": False, "error": "invalid_callback_data"}
+    agent = AGENT_REGISTRY.get("insta")() if callable(AGENT_REGISTRY.get("insta")) else AGENT_REGISTRY.get("insta")
+    return await agent.on_callback(f"issue_{verb}", {"slate_id": slate_id})
+```
+
+> `_handle_insta_render`가 에이전트를 얻는 정확한 방식(레지스트리/팩토리)을 그대로 복사할 것. 위 `AGENT_REGISTRY` 줄은 그 방식으로 대체한다.
+
+- [ ] **Step 3: import sanity + 수동 점검**
+
+Run: `cd agent-office && PYTHONPATH=.. python -c "from app.telegram import webhook; print('OK')"`
+Expected: OK.
+
+- [ ] **Step 4: 커밋**
+```bash
+git add agent-office/app/telegram/webhook.py
+git commit -m "feat(agent-office): issue_* 텔레그램 콜백 디스패치
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+```
+
+---
+
+## Task 9: 문서 + 배포 + 검증
+
+**Files:**
+- Modify: `web-backend/CLAUDE.md`, `memory/service_insta.md`
+
+- [ ] **Step 1: CLAUDE.md insta API 목록에 2개 추가**
+
+`### insta-lab` API 표에 추가:
+```
+| GET | `/api/insta/keywords/ranked` | 4신호 선별 점수 + eligible (자율 발급용) |
+| POST | `/api/insta/slates/{id}/decision` | 승인/반려 (approved→published) |
+```
+
+- [ ] **Step 2: 전체 테스트 회귀**
+
+Run:
+```bash
+cd insta-lab && PYTHONPATH=.. python -m pytest tests/ app/test_package_api.py -q
+cd ../agent-office && PYTHONPATH=.. python -m pytest tests/ -q
+```
+Expected: 모두 PASS (사전존재 stale 제외).
+
+- [ ] **Step 3: 커밋 + push (NAS 배포)**
+```bash
+cd C:/Users/jaeoh/Desktop/workspace/web-backend
+git add CLAUDE.md docs/superpowers/plans/2026-06-11-insta-autonomous-card-issuance.md
+git commit -m "docs(insta): 자율 발급 API 문서 + 구현 계획
+
+Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
+git push origin main
+```
+
+- [ ] **Step 4: 활성화 + 프로덕션 검증**
+
+배포 완료 후 (deployer rebuild ~3분):
+```bash
+# autonomous_issue 켜기 (agent_config custom_config)
+curl -X PUT https://gahusb.synology.me/api/agent-office/agents/insta \
+  -H "Content-Type: application/json" \
+  -d '{"custom_config": {"autonomous_issue": true, "select_threshold": 0.6, "max_per_day": 2}}'
+
+# 수동 트리거 대신 ranked 직접 확인
+curl -s "https://gahusb.synology.me/api/insta/keywords/ranked?threshold=0.0&limit=5" | python -m json.tool
+```
+Expected: ranked 응답에 `final_score`/`eligible`/`breakdown`. 09:30 cron 또는 수동 command로 프리뷰가 텔레그램에 도착하는지 확인.
+
+- [ ] **Step 5: 메모리 갱신**
+
+`memory/service_insta.md`에 자율 발급 파이프라인(4신호 선별·승인 게이트·상태머신) 추가 + 스마트에이전트 3종 완료 표시.
+
+---
+
+## Self-Review
+
+**Spec coverage:**
+- 선별 4신호 → Task 2(freshness/account_fit/dedup) + Task 3(claude). ✓
+- threshold 게이트 0~N → Task 2 + Task 6(max_per_day). ✓
+- 승인 게이트 + 콜백 → Task 6(프리뷰) + Task 7(approve/reject/regen) + Task 8(디스패치). ✓
+- 상태머신 + 발행이력 → Task 1. ✓
+- 하위호환(autonomous_issue=false) → Task 6 Step 3(기존 블록 유지). ✓
+- graceful Claude 실패 → Task 3(빈 dict) + Task 2(renormalize). ✓
+- 성과지표 제외(YAGNI) → 계획에 없음. ✓
+
+**Placeholder scan:** 모든 코드 스텝에 실제 코드 포함. 단 Task 5/8의 "기존 변수명/에이전트 해석 방식 모방"은 실제 파일 확인을 요구하는 의도적 지시(해당 파일이 코드 소유) — placeholder 아님.
+
+**Type consistency:** `score_candidates(candidates, issued_topics, prefs, claude_scores, weights, threshold, now_iso)` Task2 정의 ↔ Task4 호출 일치. `set_slate_decision(slate_id, decision)` Task1 ↔ Task4 일치. `insta_ranked(threshold, limit)`/`insta_decision(slate_id, decision)` Task5 ↔ Task6/7 일치. 콜백 액션 `issue_approve/issue_reject/issue_regen` Task7 ↔ Task8 prefix 파싱 일치. `_generate_and_preview(pick)` Task6 정의 ↔ Task7(regen) 호출 일치.

docs/superpowers/specs/2026-06-11-insta-autonomous-card-issuance-design.md

@@ -0,0 +1,120 @@
+# insta 자율 카드 발급 (스마트 에이전트 3번) — 설계
+
+> 작성 2026-06-11. InstaAgent를 "후보 푸시/단순 auto_select"에서 **선별 지능 + 승인 게이트 + 카덴스/추적**을 갖춘 자율 발급 파이프라인으로 확장.
+
+## 1. 목표
+
+매일 09:30, InstaAgent가 **발행할 가치 있는 주제만 자율 선별**해 카드를 생성·렌더하고, **카드별 승인 게이트**로 사람이 최종 결정(브랜드 안전)한 뒤 업로드용 카드를 발급한다. 발행 상태·이력을 추적해 중복 회피·카덴스 판단에 환류한다.
+
+Instagram Graph API는 사용하지 않는다(수동 업로드). "발행(published)" = 승인되어 업로드 준비가 끝난 카드 상태 + 텔레그램으로 전달.
+
+## 2. 현재 상태 (배경)
+
+- insta-lab: 뉴스수집→키워드추출→슬레이트 생성(`POST /slates`)→Redis push→Windows 워커 렌더→webhook이 `card_assets` 등록. (2026-06-11 렌더 갭 복구 완료, slate 상태 `draft→rendered`.)
+- agent-office `InstaAgent`: 09:30 cron에서 collect+extract 후 (기본) 텔레그램 후보 버튼 푸시 / (`auto_select=True`) 카테고리 1위 키워드 자동 렌더+미디어그룹 발송. 버튼 탭 → `render_{kid}` 콜백 → 슬레이트 생성·렌더·발송.
+- `account_preferences`(카테고리 가중치) 존재. 발행 성과 추적은 없음.
+
+즉 "생성→렌더→전달"은 동작한다. 본 설계는 그 앞단의 **자율 선별**과 뒷단의 **승인·추적**을 추가한다.
+
+## 3. 요구사항 (확정)
+
+- **선별 신호 4종**: ① 중복 회피(최근 발행/반려 주제 제외) ② 신선도(뉴스 최신성) ③ 계정 컨셉 적합도(카테고리 가중치) ④ Claude 판단(카드가치·흥미·리스크). 가중합 → threshold 게이트.
+- **카덴스**: 에이전트 결정 — 매일 09:30, threshold 이상인 픽만 `max_per_day`까지(0~N 가변). 가치 없으면 발행 안 함.
+- **승인**: 카드별 게이트. 자동 생성 후 텔레그램 프리뷰 `[✅승인][❌반려][🔄재생성]`. 승인만 published.
+- **추적**: slate 상태 ∈ `{draft, rendered, rejected, published}` + 발행 이력. decision=approved→`published`, decision=rejected→`rejected`("approved"는 별도 저장 상태가 아니라 decision 액션). 성과 지표(좋아요·도달)는 범위 외(YAGNI — IG API 없어 수동).
+
+## 4. 아키텍처 (접근법 A: 데이터 있는 곳에서 선별, 에이전트는 오케스트레이션)
+
+```
+[09:30 cron] InstaAgent.on_schedule (autonomous_issue=True)
+ 1. collect + extract                                  (기존 재사용)
+ 2. GET /api/insta/keywords/ranked?threshold&limit     ← insta-lab: 4신호 점수
+ 3. eligible 픽마다(max_per_day): create_slate → wait render  (기존 재사용)
+ 4. 텔레그램 프리뷰(커버1장+요약) + [✅][❌][🔄] + agent_task(requires_approval) → waiting
+
+[telegram webhook] → InstaAgent.on_callback
+   issue_approve_{id} → POST /slates/{id}/decision{approved} → published + 10장 미디어그룹 + /package zip
+   issue_reject_{id}  → POST /slates/{id}/decision{rejected}
+   issue_regen_{id}   → 같은 키워드로 슬레이트 재생성(새 카피) → 새 프리뷰 (이전 슬레이트 폐기)
+```
+
+경계: **insta-lab = 선별 점수 + 상태머신(DB 소유)**, **agent-office = cron 오케스트레이션 + 텔레그램 승인**.
+
+## 5. insta-lab 상세
+
+### 5.1 `app/selection.py` (순수 함수)
+입력: 후보 키워드 리스트, 발행/반려 이력, 카테고리 선호 가중치, (선택) Claude 판단 점수.
+출력: 후보별 `{keyword_id, final_score, breakdown:{dedup,freshness,account_fit,claude}, eligible}`.
+
+신호별 정의:
+- **dedup** (0 또는 1, exclude 게이트): 최근 `dedup_window_days`(기본 14) 내 `published`/`rejected` 슬레이트와 동일 키워드(정규화 후 exact/substring) + 동일 카테고리면 `eligible=False`로 제외.
+- **freshness** (0~1): 키워드 `suggested_at`이 최근일수록 높음(예: 24h=1.0, 선형 감쇠, 7일+=0).
+- **account_fit** (0~1): `account_preferences[category].weight`(정규화) × 키워드 자체 score.
+- **claude** (0~1): Claude Haiku가 후보 일괄 평가(아래 5.3). 실패 시 이 항 제외하고 나머지로 정규화(graceful).
+- **final_score** = 가중합 `w_fresh*freshness + w_fit*account_fit + w_claude*claude` (dedup 제외 통과한 것만). 기본 가중치 `{fresh:0.3, fit:0.3, claude:0.4}`. `eligible = (dedup 통과) and (final_score >= threshold)`.
+
+### 5.2 엔드포인트
+- `GET /api/insta/keywords/ranked?limit=N&threshold=T`
+  - 내부에서: 미사용 키워드 조회 + 발행/반려 이력 조회 + 선호 조회 + Claude 일괄 호출 → `selection.py` → 정렬된 후보 + breakdown + `eligible` 반환.
+- `POST /api/insta/slates/{id}/decision` body `{"decision": "approved"|"rejected"}`
+  - approved → `status='published'`, `published_at=now`, `decision_at=now` (멱등: 이미 published면 no-op).
+  - rejected → `status='rejected'`, `decision_at=now`.
+
+### 5.3 Claude 판단 프롬프트 (insta-lab, 기존 ANTHROPIC 클라이언트 재사용)
+- 1회 호출로 후보 N개 일괄 평가. 입력: 각 후보 `{keyword, category}`. 출력: JSON `[{keyword_id, score(0~1), reason}]`.
+- 기준: 카드뉴스로 만들 가치(흥미·시의성·정보성) 및 리스크(민감·논란). 모델 `ANTHROPIC_MODEL_HAIKU`.
+- 실패/파싱오류 → 빈 결과 반환 → selection이 claude 항 제외.
+
+### 5.4 스키마 (idempotent ALTER)
+- `card_slates`에 `published_at TEXT NULL`, `decision_at TEXT NULL` 추가.
+- 상태값: `draft → rendered → approved/rejected → published`. (approved는 과도기 상태 없이 decision=approved 시 바로 published로 둔다 — 단순화. rejected는 종결.)
+- 발행 이력 = `SELECT keyword, category, published_at FROM card_slates WHERE status IN ('published','rejected') AND COALESCE(published_at,decision_at) >= datetime('now', '-D days')`.
+
+## 6. agent-office 상세
+
+### 6.1 `InstaAgent.on_schedule`
+- `custom_config.autonomous_issue` 분기. False면 **기존 동작 유지**(candidate-push / auto_select) — 하위호환.
+- True면: collect+extract(기존) → `service_proxy.insta_ranked(threshold, limit=max_per_day)` → `eligible` 픽 순회(최대 `max_per_day`):
+  - 슬레이트 생성·렌더 대기(기존 `_render_and_push`의 생성·대기 부분 재사용/분리) → **프리뷰 발송**(6.3) → `create_task(requires_approval=True)` → `waiting` 상태.
+- eligible 0개 → "오늘 발행할 가치 있는 주제 없음" 1통.
+
+### 6.2 콜백 (telegram webhook → `on_callback`)
+- `issue_approve_{slate_id}`: `insta_decision(slate_id, "approved")` → 전체 10장 미디어그룹 + `/package` zip 전달 + "✅ 발행 완료" → 해당 task succeeded.
+- `issue_reject_{slate_id}`: `insta_decision(slate_id, "rejected")` → "❌ 반려됨" → task 종료.
+- `issue_regen_{slate_id}`: 해당 슬레이트의 키워드로 새 슬레이트 생성(새 Claude 카피)·렌더 → 새 프리뷰. 이전 슬레이트는 rejected 처리.
+
+### 6.3 텔레그램 프리뷰 (미디어그룹은 인라인 키보드 불가)
+- 커버(01.png) 단장 사진 + 캡션: 키워드·카테고리·`final_score`·breakdown 요약 + inline `[✅승인][❌반려][🔄재생성]` (`callback_data=issue_*_{slate_id}`).
+
+### 6.4 설정 (`agent_config.custom_config`)
+- `autonomous_issue` (bool, 기본 false), `select_threshold` (기본 0.6), `max_per_day` (기본 2), `dedup_window_days` (기본 14).
+
+### 6.5 service_proxy 추가
+- `insta_ranked(threshold, limit)` → `GET /keywords/ranked`
+- `insta_decision(slate_id, decision)` → `POST /slates/{id}/decision`
+
+## 7. 에러 처리 / 엣지
+
+- ranked의 Claude 실패 → 룰 점수만으로 진행(graceful), 경고 로그.
+- eligible 0개 → 안내 1통(또는 무음 옵션, 기본 안내).
+- 렌더 실패 → task failed 통지, 프리뷰 미발송.
+- 승인 미응답 → 슬레이트 pending(rendered) 유지, 자동 발행 안 함(안전). 만료 없음.
+- 멱등: 중복 승인/반려 no-op. cron 재실행 시 이미 발행/반려 주제는 dedup으로 회피.
+- regen 무한루프 방지: regen은 사용자 트리거(버튼)라 자동 반복 없음.
+
+## 8. 테스트
+
+- **insta-lab**: `selection.py` 순수 단위테스트(dedup 최근 제외 / freshness 정렬 / account_fit 가중 / 가중합·threshold 게이트 / claude 실패 시 정규화). ranked 엔드포인트(Claude mock). decision 엔드포인트(approved→published+published_at, rejected, 멱등).
+- **agent-office**: 자율 `on_schedule`(proxy mock: ranked eligible→슬레이트 생성→프리뷰 발송 + task requires_approval). 콜백 approve/reject/regen(proxy·messaging mock).
+
+## 9. 범위 외 (YAGNI)
+
+- 발행 성과 지표(좋아요·도달) 수집/학습 — IG API 미사용, 수동 입력 부담으로 제외.
+- 신뢰도 하이브리드 자동발행(승인 생략) — 승인 게이트로 통일.
+- 임베딩 기반 유사도 dedup — 정규화 exact/substring + 카테고리로 충분(추후 필요 시 확장).
+
+## 10. 영향받는 파일
+
+- insta-lab: `app/selection.py`(신규), `app/main.py`(ranked·decision 라우트), `app/db.py`(컬럼 ALTER + 발행이력/상태 헬퍼), `tests/`.
+- agent-office: `app/agents/insta.py`(자율 경로·콜백), `app/service_proxy.py`(2 헬퍼), `app/webhook.py`(issue_* 콜백 디스패치), `tests/`.
+- web-backend/CLAUDE.md insta API 목록 + `service_insta.md` 메모리 갱신.

insta-lab/app/db.py

@@ -124,6 +124,13 @@ def init_db() -> None:
                     (cat, 1.0),
                 )
 
+        # 발행 상태 컬럼 (idempotent ALTER) — 자율 발급 파이프라인
+        cs_cols = [r[1] for r in conn.execute("PRAGMA table_info(card_slates)").fetchall()]
+        if "published_at" not in cs_cols:
+            conn.execute("ALTER TABLE card_slates ADD COLUMN published_at TEXT")
+        if "decision_at" not in cs_cols:
+            conn.execute("ALTER TABLE card_slates ADD COLUMN decision_at TEXT")
+
 
 # ── news_articles ────────────────────────────────────────────────
 def add_news_article(row: Dict[str, Any]) -> int:
@@ -217,6 +224,39 @@ def update_slate_status(slate_id: int, status: str) -> None:
         )
 
 
+def set_slate_decision(slate_id: int, decision: str) -> None:
+    """승인/반려 결정 기록. approved→published(+published_at), rejected→rejected.
+    멱등: 이미 published면 published_at 유지."""
+    now = "strftime('%Y-%m-%dT%H:%M:%fZ','now')"
+    with _conn() as conn:
+        if decision == "approved":
+            conn.execute(
+                f"UPDATE card_slates SET status='published', "
+                f"published_at=COALESCE(published_at, {now}), decision_at={now} "
+                f"WHERE id=?",
+                (slate_id,),
+            )
+        elif decision == "rejected":
+            conn.execute(
+                f"UPDATE card_slates SET status='rejected', decision_at={now} WHERE id=?",
+                (slate_id,),
+            )
+        else:
+            raise ValueError(f"invalid decision: {decision}")
+
+
+def list_recent_issued_topics(window_days: int = 14) -> List[Dict[str, Any]]:
+    """최근 window_days 내 published/rejected 슬레이트의 (keyword, category). dedup용."""
+    with _conn() as conn:
+        rows = conn.execute(
+            "SELECT keyword, category FROM card_slates "
+            "WHERE status IN ('published','rejected') "
+            "AND COALESCE(published_at, decision_at) >= datetime('now', ?)",
+            (f"-{int(window_days)} days",),
+        ).fetchall()
+    return [dict(r) for r in rows]
+
+
 def get_card_slate(slate_id: int) -> Optional[Dict[str, Any]]:
     with _conn() as conn:
         row = conn.execute("SELECT * FROM card_slates WHERE id=?", (slate_id,)).fetchone()

insta-lab/app/main.py

@@ -6,6 +6,7 @@ import json
 import logging
 import os
 import zipfile
+from datetime import datetime, timezone
 from typing import Optional
 
 from fastapi import FastAPI, HTTPException, BackgroundTasks, Body, Query
@@ -21,7 +22,7 @@ from .config import (
 )
 import redis.asyncio as aioredis
 
-from . import db, news_collector, keyword_extractor, card_writer, trend_collector
+from . import db, news_collector, keyword_extractor, card_writer, trend_collector, selection, selection_judge
 from .internal_router import router as internal_router
 
 logger = logging.getLogger(__name__)
@@ -152,6 +153,26 @@ def list_keywords(
     return {"items": db.list_trending_keywords(category=category, used=used)}
 
 
+@app.get("/api/insta/keywords/ranked")
+def ranked_keywords(
+    limit: int = Query(20, ge=1, le=100),
+    threshold: float = Query(0.6, ge=0.0, le=1.0),
+    dedup_window_days: int = Query(14, ge=1, le=90),
+):
+    candidates = db.list_trending_keywords(used=False)
+    if not candidates:
+        return {"items": []}
+    issued = db.list_recent_issued_topics(window_days=dedup_window_days)
+    prefs = {p["category"]: p["weight"] for p in db.get_preferences()}
+    claude_scores = selection_judge.judge_candidates(candidates)
+    now_iso = datetime.now(timezone.utc).isoformat()
+    scored = selection.score_candidates(
+        candidates, issued, prefs, claude_scores=claude_scores,
+        threshold=threshold, now_iso=now_iso,
+    )
+    return {"items": scored[:limit]}
+
+
 # ── Slates ───────────────────────────────────────────────────────
 class SlateRequest(BaseModel):
     keyword: str
@@ -282,6 +303,20 @@ def download_package(slate_id: int):
     })
 
 
+class DecisionBody(BaseModel):
+    decision: str  # "approved" | "rejected"
+
+
+@app.post("/api/insta/slates/{slate_id}/decision")
+def slate_decision(slate_id: int, body: DecisionBody):
+    if not db.get_card_slate(slate_id):
+        raise HTTPException(404, "slate not found")
+    if body.decision not in ("approved", "rejected"):
+        raise HTTPException(400, "decision must be approved|rejected")
+    db.set_slate_decision(slate_id, body.decision)
+    return db.get_card_slate(slate_id)
+
+
 @app.delete("/api/insta/slates/{slate_id}")
 def delete_slate(slate_id: int):
     if not db.get_card_slate(slate_id):

insta-lab/app/selection.py

@@ -0,0 +1,83 @@
+"""발행 가치 자율 선별 — 순수 점수 함수 (외부 IO 없음, 단위테스트 대상).
+
+신호: dedup(게이트), freshness, account_fit, claude(선택).
+final = 가중합(존재하는 신호만 정규화). eligible = dedup통과 and final>=threshold.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+DEFAULT_WEIGHTS = {"freshness": 0.3, "account_fit": 0.3, "claude": 0.4}
+FRESH_WINDOW_HOURS = 168.0  # 7일 → 0
+
+
+def _parse_iso(s: str) -> datetime:
+    return datetime.fromisoformat(s.replace("Z", "+00:00")).astimezone(timezone.utc)
+
+
+def _norm(kw: str) -> str:
+    return (kw or "").strip().lower()
+
+
+def _is_duplicate(keyword: str, category: str, issued: List[Dict[str, Any]]) -> bool:
+    n = _norm(keyword)
+    if not n:
+        return False
+    for it in issued:
+        if it.get("category") != category:
+            continue
+        m = _norm(it.get("keyword", ""))
+        if not m:
+            continue
+        if n == m or n in m or m in n:
+            return True
+    return False
+
+
+def _freshness(suggested_at: str, now: datetime) -> float:
+    try:
+        hours = (now - _parse_iso(suggested_at)).total_seconds() / 3600.0
+    except Exception:
+        return 0.0
+    return max(0.0, min(1.0, 1.0 - hours / FRESH_WINDOW_HOURS))
+
+
+def score_candidates(
+    candidates: List[Dict[str, Any]],
+    issued_topics: List[Dict[str, Any]],
+    prefs: Dict[str, float],
+    claude_scores: Optional[Dict[int, float]] = None,
+    weights: Optional[Dict[str, float]] = None,
+    threshold: float = 0.6,
+    now_iso: Optional[str] = None,
+) -> List[Dict[str, Any]]:
+    w = weights or DEFAULT_WEIGHTS
+    now = _parse_iso(now_iso) if now_iso else datetime.now(timezone.utc)
+    max_w = max(prefs.values()) if prefs else 1.0
+    if max_w <= 0:
+        max_w = 1.0
+    out: List[Dict[str, Any]] = []
+    for c in candidates:
+        cat = c.get("category", "")
+        dup = _is_duplicate(c.get("keyword", ""), cat, issued_topics)
+        freshness = _freshness(c.get("suggested_at", ""), now)
+        weight = prefs.get(cat, 1.0)
+        account_fit = max(0.0, min(1.0, (weight / max_w) * float(c.get("score", 0.0))))
+        claude = None
+        if claude_scores is not None and c["id"] in claude_scores:
+            claude = max(0.0, min(1.0, float(claude_scores[c["id"]])))
+        parts = [("freshness", freshness), ("account_fit", account_fit)]
+        if claude is not None:
+            parts.append(("claude", claude))
+        total_w = sum(w[name] for name, _ in parts)
+        final = sum(w[name] * val for name, val in parts) / total_w if total_w else 0.0
+        eligible = (not dup) and (final >= threshold)
+        out.append({
+            "id": c["id"], "keyword": c.get("keyword"), "category": cat,
+            "final_score": round(final, 4), "eligible": eligible,
+            "breakdown": {"dedup_excluded": dup, "freshness": round(freshness, 4),
+                          "account_fit": round(account_fit, 4), "claude": claude},
+        })
+    out.sort(key=lambda x: (-x["eligible"], -x["final_score"]))
+    return out

insta-lab/app/selection_judge.py

@@ -0,0 +1,52 @@
+"""Claude Haiku 일괄 카드가치 판단. 실패/미설정 시 빈 dict (graceful)."""
+from __future__ import annotations
+
+import json
+import logging
+import re
+from typing import Any, Dict, List
+
+from anthropic import Anthropic
+
+from .config import ANTHROPIC_API_KEY, ANTHROPIC_MODEL_HAIKU
+
+logger = logging.getLogger(__name__)
+
+PROMPT = """다음 인스타 카드뉴스 후보 키워드들을 카드로 만들 가치(흥미·시의성·정보성)와
+리스크(민감·논란)를 종합해 0~1 점수로 평가해라. 코드펜스 없이 JSON 배열로만 출력:
+[{{"keyword_id": <id>, "score": <0~1>}}, ...]
+
+후보:
+{items}"""
+
+
+def _strip_codefence(s: str) -> str:
+    s = s.strip()
+    if s.startswith("```"):
+        s = re.sub(r"^```(?:json)?\s*|\s*```$", "", s).strip()
+    return s
+
+
+def parse_judge_response(raw: str) -> Dict[int, float]:
+    try:
+        data = json.loads(_strip_codefence(raw))
+        return {int(d["keyword_id"]): float(d["score"]) for d in data}
+    except Exception:
+        logger.warning("judge 응답 파싱 실패")
+        return {}
+
+
+def judge_candidates(candidates: List[Dict[str, Any]]) -> Dict[int, float]:
+    if not ANTHROPIC_API_KEY or not candidates:
+        return {}
+    items = "\n".join(f'- id={c["id"]}: {c["keyword"]} ({c["category"]})' for c in candidates)
+    try:
+        client = Anthropic(api_key=ANTHROPIC_API_KEY)
+        resp = client.messages.create(
+            model=ANTHROPIC_MODEL_HAIKU, max_tokens=1024,
+            messages=[{"role": "user", "content": PROMPT.format(items=items)}],
+        )
+        return parse_judge_response(resp.content[0].text)
+    except Exception:
+        logger.exception("judge_candidates 호출 실패")
+        return {}

insta-lab/tests/test_db_decision.py

@@ -0,0 +1,47 @@
+import os
+import pytest
+from app import db, config
+
+
+@pytest.fixture
+def fresh_db(tmp_path, monkeypatch):
+    monkeypatch.setattr(config, "DB_PATH", str(tmp_path / "insta.db"))
+    monkeypatch.setattr(db, "DB_PATH", str(tmp_path / "insta.db"))
+    db.init_db()
+
+
+def test_set_slate_decision_approved_publishes(fresh_db):
+    sid = db.add_card_slate({"keyword": "금리", "category": "economy"})
+    db.set_slate_decision(sid, "approved")
+    s = db.get_card_slate(sid)
+    assert s["status"] == "published"
+    assert s["published_at"] is not None
+    assert s["decision_at"] is not None
+
+
+def test_set_slate_decision_rejected(fresh_db):
+    sid = db.add_card_slate({"keyword": "환율", "category": "economy"})
+    db.set_slate_decision(sid, "rejected")
+    s = db.get_card_slate(sid)
+    assert s["status"] == "rejected"
+    assert s["decision_at"] is not None
+    assert s["published_at"] is None
+
+
+def test_set_slate_decision_idempotent(fresh_db):
+    sid = db.add_card_slate({"keyword": "주식", "category": "economy"})
+    db.set_slate_decision(sid, "approved")
+    first = db.get_card_slate(sid)["published_at"]
+    db.set_slate_decision(sid, "approved")
+    assert db.get_card_slate(sid)["published_at"] == first
+
+
+def test_list_recent_issued_topics(fresh_db):
+    a = db.add_card_slate({"keyword": "금리", "category": "economy"})
+    b = db.add_card_slate({"keyword": "우울증", "category": "psychology"})
+    db.set_slate_decision(a, "approved")
+    db.set_slate_decision(b, "rejected")
+    topics = db.list_recent_issued_topics(window_days=14)
+    pairs = {(t["keyword"], t["category"]) for t in topics}
+    assert ("금리", "economy") in pairs
+    assert ("우울증", "psychology") in pairs

insta-lab/tests/test_ranked_decision_api.py

@@ -0,0 +1,78 @@
+import pytest
+from fastapi.testclient import TestClient
+from app import db, config, selection_judge
+
+
+@pytest.fixture
+def client(tmp_path, monkeypatch):
+    monkeypatch.setattr(config, "DB_PATH", str(tmp_path / "insta.db"))
+    monkeypatch.setattr(db, "DB_PATH", str(tmp_path / "insta.db"))
+    monkeypatch.setattr(selection_judge, "judge_candidates", lambda c: {})
+    db.init_db()
+    from app.main import app
+    return TestClient(app)
+
+
+def test_ranked_returns_sorted_eligible(client, monkeypatch):
+    db.add_trending_keyword({"keyword": "금리", "category": "economy", "score": 0.9})
+    r = client.get("/api/insta/keywords/ranked?threshold=0.0&limit=10")
+    assert r.status_code == 200
+    items = r.json()["items"]
+    assert len(items) >= 1
+    assert "final_score" in items[0] and "eligible" in items[0]
+
+
+def test_decision_approve_publishes(client):
+    sid = db.add_card_slate({"keyword": "금리", "category": "economy"})
+    r = client.post(f"/api/insta/slates/{sid}/decision", json={"decision": "approved"})
+    assert r.status_code == 200
+    assert db.get_card_slate(sid)["status"] == "published"
+
+
+def test_decision_reject(client):
+    sid = db.add_card_slate({"keyword": "환율", "category": "economy"})
+    r = client.post(f"/api/insta/slates/{sid}/decision", json={"decision": "rejected"})
+    assert r.status_code == 200
+    assert db.get_card_slate(sid)["status"] == "rejected"
+
+
+def test_decision_invalid_400(client):
+    sid = db.add_card_slate({"keyword": "x", "category": "economy"})
+    r = client.post(f"/api/insta/slates/{sid}/decision", json={"decision": "maybe"})
+    assert r.status_code == 400
+
+
+def test_decision_unknown_slate_404(client):
+    r = client.post("/api/insta/slates/99999/decision", json={"decision": "approved"})
+    assert r.status_code == 404
+
+
+def test_ranked_respects_dedup_window(client):
+    """dedup_window_days param이 list_recent_issued_topics window에 반영되는지 검증.
+
+    '금리' 키워드를 방금 approved(published) 상태로 기록한 뒤:
+    - dedup_window_days=30 → 방금 발행 = window 안 → eligible False
+    - dedup_window_days=1  → DB datetime이 정각 경계 직전이라도 여전히 1일 안이므로 eligible False
+      (확인: 반드시 eligible=False)
+    추가로 두 번째 키워드(word2)는 아직 발행 이력 없으므로 window 무관하게 eligible True.
+    """
+    # 방금 발행된 키워드 등록 + 슬레이트 approved 처리
+    db.add_trending_keyword({"keyword": "금리", "category": "economy", "score": 0.9})
+    sid = db.add_card_slate({"keyword": "금리", "category": "economy"})
+    db.set_slate_decision(sid, "approved")  # published_at = now
+
+    # 발행 이력 없는 키워드 추가
+    db.add_trending_keyword({"keyword": "환율", "category": "economy", "score": 0.8})
+
+    # window=30 → '금리'는 최근 발행이라 dedup 대상 → eligible False
+    r = client.get("/api/insta/keywords/ranked?threshold=0.0&limit=10&dedup_window_days=30")
+    assert r.status_code == 200
+    items = r.json()["items"]
+    keumni = next((i for i in items if i["keyword"] == "금리"), None)
+    assert keumni is not None, "'금리' 항목이 ranked 응답에 없음"
+    assert keumni["eligible"] is False, "dedup_window_days=30 내 발행 → eligible은 False여야 함"
+
+    # 발행 이력 없는 '환율'은 어떤 window에서도 eligible True
+    hwanul = next((i for i in items if i["keyword"] == "환율"), None)
+    assert hwanul is not None, "'환율' 항목이 ranked 응답에 없음"
+    assert hwanul["eligible"] is True, "발행 이력 없는 키워드는 eligible True여야 함"

insta-lab/tests/test_selection.py

@@ -0,0 +1,55 @@
+from app.selection import score_candidates
+
+NOW = "2026-06-11T00:00:00Z"
+
+
+def _cand(kid, kw, cat, score, suggested_at):
+    return {"id": kid, "keyword": kw, "category": cat, "score": score, "suggested_at": suggested_at}
+
+
+def test_dedup_excludes_recent_issued():
+    cands = [_cand(1, "금리", "economy", 0.9, "2026-06-11T00:00:00Z")]
+    issued = [{"keyword": "금리", "category": "economy"}]
+    out = score_candidates(cands, issued, prefs={}, claude_scores=None, threshold=0.0, now_iso=NOW)
+    assert out[0]["eligible"] is False
+
+
+def test_freshness_recent_higher():
+    fresh = _cand(1, "A", "economy", 0.5, "2026-06-11T00:00:00Z")
+    stale = _cand(2, "B", "economy", 0.5, "2026-06-04T00:00:00Z")
+    out = {c["id"]: c for c in score_candidates([fresh, stale], [], {}, None, threshold=0.0, now_iso=NOW)}
+    assert out[1]["breakdown"]["freshness"] > out[2]["breakdown"]["freshness"]
+
+
+def test_account_fit_uses_weight():
+    cands = [_cand(1, "A", "economy", 0.8, NOW), _cand(2, "B", "psychology", 0.8, NOW)]
+    prefs = {"economy": 2.0, "psychology": 1.0}
+    out = {c["id"]: c for c in score_candidates(cands, [], prefs, None, threshold=0.0, now_iso=NOW)}
+    assert out[1]["breakdown"]["account_fit"] > out[2]["breakdown"]["account_fit"]
+
+
+def test_threshold_gate():
+    cands = [_cand(1, "A", "economy", 0.1, "2026-06-01T00:00:00Z")]
+    out = score_candidates(cands, [], {}, None, threshold=0.6, now_iso=NOW)
+    assert out[0]["eligible"] is False
+
+
+def test_claude_missing_renormalizes():
+    cands = [_cand(1, "A", "economy", 1.0, NOW)]
+    out = score_candidates(cands, [], {"economy": 1.0}, None, threshold=0.0, now_iso=NOW)
+    assert out[0]["breakdown"]["claude"] is None
+    assert 0.0 <= out[0]["final_score"] <= 1.0
+
+
+def test_claude_included_when_provided():
+    cands = [_cand(1, "A", "economy", 0.5, NOW)]
+    out = score_candidates(cands, [], {"economy": 1.0}, {1: 1.0}, threshold=0.0, now_iso=NOW)
+    assert out[0]["breakdown"]["claude"] == 1.0
+
+
+def test_all_zero_prefs_no_crash():
+    cands = [{"id": 1, "keyword": "A", "category": "economy", "score": 0.8,
+              "suggested_at": "2026-06-11T00:00:00Z"}]
+    prefs = {"economy": 0.0, "psychology": 0.0}
+    out = score_candidates(cands, [], prefs, None, threshold=0.0, now_iso="2026-06-11T00:00:00Z")
+    assert out[0]["breakdown"]["account_fit"] == 0.0  # 0가중 → fit 0, 크래시 없음

insta-lab/tests/test_selection_judge.py

@@ -0,0 +1,20 @@
+from app import selection_judge
+
+
+def test_parse_judge_response_ok():
+    raw = '[{"keyword_id": 1, "score": 0.8}, {"keyword_id": 2, "score": 0.3}]'
+    assert selection_judge.parse_judge_response(raw) == {1: 0.8, 2: 0.3}
+
+
+def test_parse_judge_response_codefence():
+    raw = '```json\n[{"keyword_id": 5, "score": 0.5}]\n```'
+    assert selection_judge.parse_judge_response(raw) == {5: 0.5}
+
+
+def test_parse_judge_response_garbage_returns_empty():
+    assert selection_judge.parse_judge_response("not json") == {}
+
+
+def test_judge_candidates_no_key_returns_empty(monkeypatch):
+    monkeypatch.setattr(selection_judge, "ANTHROPIC_API_KEY", "")
+    assert selection_judge.judge_candidates([{"id": 1, "keyword": "x", "category": "economy"}]) == {}