diff --git a/docs/superpowers/plans/2026-06-11-insta-autonomous-card-issuance.md b/docs/superpowers/plans/2026-06-11-insta-autonomous-card-issuance.md new file mode 100644 index 0000000..bc681fd --- /dev/null +++ b/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) " +``` + +--- + +## 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) " +``` + +--- + +## 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": , "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) " +``` + +--- + +## 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) " +``` + +--- + +## 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) " +``` + +--- + +## 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"🎴 {pick['keyword']} ({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) " +``` + +--- + +## 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) " +``` + +--- + +## 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) " +``` + +--- + +## 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) " +``` + +--- + +## 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) " +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) 호출 일치.