Files

gahusb 4d02d9c321 docs(plan): insta 자율 카드 발급 구현 계획 (9 tasks, TDD)

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

2026-06-11 02:11:32 +09:00

38 KiB

Raw Blame History

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:

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 직후)에 추가:

        # 발행 상태 컬럼 (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 아래)에 추가:

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: 커밋

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:

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:

"""발행 가치 자율 선별 — 순수 점수 함수 (외부 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: 커밋

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:

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:

"""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: 커밋

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:

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 블록에 추가:

from datetime import datetime, timezone
from . import selection, selection_judge

list_keywords 엔드포인트 아래에 추가:

@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 위)에 추가:

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 회귀 + 커밋

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 변수명을 동일하게 사용:

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: 커밋

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:

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 읽은 직후:

        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() 다음의 분기를 교체:

            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에 추가 — 슬레이트 생성·렌더(기존 흐름) 후 커버 프리뷰 발송:

    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: 커밋

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 사용)을 따라 추가:

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: 커밋

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에 추가:

@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을 확장:

    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: 커밋

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_ 분기 아래에 추가:

    if callback_id.startswith("issue_"):
        return await _handle_insta_issue(callback_query, callback_id)

Step 2: _handle_insta_issue 추가

_handle_insta_render(103행~)를 본떠 추가:

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: 커밋

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:

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 배포)

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분):

# 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) 호출 일치.

38 KiB Raw Blame History

insta 자율 카드 발급 (스마트 에이전트 3번) Implementation Plan

File Structure

Task 1: insta-lab DB — 발행 상태 컬럼 + 헬퍼

Task 2: insta-lab — selection.py 순수 점수

Task 3: insta-lab — Claude 카드가치 판단 (selection_judge.py)

Task 4: insta-lab — GET /api/insta/keywords/ranked

Task 5: agent-office — service_proxy 헬퍼

Task 6: agent-office — InstaAgent 자율 발급 경로 + 프리뷰

Task 6.5: agent-office — messaging.send_photo (없을 경우만)

Task 7: agent-office — issue_* 콜백 처리

Task 8: agent-office — 텔레그램 콜백 디스패치

Task 9: 문서 + 배포 + 검증

Self-Review

38 KiB

Raw Blame History

Task 2: insta-lab — `selection.py` 순수 점수

Task 3: insta-lab — Claude 카드가치 판단 (`selection_judge.py`)

Task 4: insta-lab — `GET /api/insta/keywords/ranked`

Task 6.5: agent-office — `messaging.send_photo` (없을 경우만)

Task 7: agent-office — `issue_*` 콜백 처리