feat(editorial): Stage 3 chat + assets 격리 + publish snapshot + thumbnail + commercial

packages/ai-server/src/bootstrap.py

@@ -96,6 +96,12 @@ async def health_check():
             }
 
     # #416 — 모든 비즈니스 routes 가 gRPC 로 옮겨짐. router include 없음.
+    # 예외: editorial article chat (#429 Stage 3) — multi-turn / admin UI 호출 패턴.
+    # #429: chat sessions / messages / article load 모두 assets staging.
+    from src.editorial_article_chat.api import create_router as create_chat_router
+
+    assets_db = application.infrastructure().assets_database_manager()
+    app.include_router(create_chat_router(assets_db))
 
     return app
 

packages/ai-server/src/config/_container.py

@@ -21,6 +21,7 @@
 from src.services.metadata.management.failed_items_manager import FailedItemsManager
 from src.services.raw_posts.adapters import build_default_adapters
 from src.managers.llm.adapters.nano_banana import NanoBananaClient
+from src.managers.llm.adapters.openai_image import OpenAIImageClient
 from src.services.raw_posts.discovery.composite_filter import CompositeFilter
 from src.services.raw_posts.pipeline import RawPostsPipeline
 from src.services.raw_posts.processors import (
@@ -75,6 +76,20 @@ class InfrastructureContainer(DeclarativeContainer):
         environment=environment,
     )
 
+    # Generic LLM image-gen adapter — raw_posts + editorial_article 공용 (#429)
+    nano_banana_client: Singleton[NanoBananaClient] = Singleton(
+        NanoBananaClient,
+        api_key=Callable(lambda env: env.GEMINI_API_KEY, environment),
+        model=Callable(lambda env: env.NANO_BANANA_MODEL, environment),
+        timeout_seconds=Callable(lambda env: env.NANO_BANANA_TIMEOUT, environment),
+    )
+
+    # OpenAI gpt-image-1 — magazine thumbnail 전용 (#429, 한글 text rendering 우수)
+    openai_image_client: Singleton[OpenAIImageClient] = Singleton(
+        OpenAIImageClient,
+        api_key=Callable(lambda env: env.OPENAI_API_KEY, environment),
+    )
+
     # #214 raw_posts_callback_client removed — ai-server writes DB directly.
 
     # ─── Postgres asyncpg pool — assets (raw_posts) vs operation (운영 DB) 분리 (#333, #369) ───
@@ -207,17 +222,10 @@ class RawPostsContainer(DeclarativeContainer):
         database_manager=infrastructure.assets_database_manager,
     )
 
-    # #348 — processor stack (Nano Banana + Gemini Pro + Gemini Flash)
-    nano_banana_client: Singleton[NanoBananaClient] = Singleton(
-        NanoBananaClient,
-        api_key=Callable(lambda env: env.GEMINI_API_KEY, environment),
-        model=Callable(lambda env: env.NANO_BANANA_MODEL, environment),
-        timeout_seconds=Callable(lambda env: env.NANO_BANANA_TIMEOUT, environment),
-    )
-
+    # #348 — processor stack uses InfrastructureContainer.nano_banana_client (#429 공유)
     hero_reframe_service: Singleton[HeroReframeService] = Singleton(
         HeroReframeService,
-        client=nano_banana_client,
+        client=infrastructure.nano_banana_client,
     )
 
     items_parser: Singleton[ItemsParser] = Singleton(
@@ -235,7 +243,7 @@ class RawPostsContainer(DeclarativeContainer):
     # #352 — composite crop + nano-banana 1:1 catalog photo per item.
     items_thumbnail_service: Singleton[ItemsThumbnailService] = Singleton(
         ItemsThumbnailService,
-        client=nano_banana_client,
+        client=infrastructure.nano_banana_client,
         r2=infrastructure.r2_client,
     )
 
@@ -344,9 +352,13 @@ class EditorialDiscoveryContainer(DeclarativeContainer):
         DependenciesContainer()
     )
 
+    # #429 architectural fix: editorial staging 은 assets DB.
+    # posts/spots/solutions read 는 별도로 operation pool 사용
+    # (gather_discovery_input 안에서 두 풀 모두 acquire).
     repository: Singleton[EditorialDiscoveryRepository] = Singleton(
         EditorialDiscoveryRepository,
-        database_manager=infrastructure.operation_database_manager,
+        assets_db=infrastructure.assets_database_manager,
+        operation_db=infrastructure.operation_database_manager,
     )
 
     scheduler: Singleton[DiscoveryScheduler] = Singleton(
@@ -365,9 +377,10 @@ class EditorialArticlePickupContainer(DeclarativeContainer):
         DependenciesContainer()
     )
 
+    # #429: assets pool — editorial_articles staging
     scheduler: Singleton[ArticlePickupScheduler] = Singleton(
         ArticlePickupScheduler,
-        database_manager=infrastructure.operation_database_manager,
+        database_manager=infrastructure.assets_database_manager,
         queue_manager=infrastructure.queue_manager,
     )
 

packages/ai-server/src/editorial_article/graph.py

@@ -10,6 +10,7 @@
 
 from .nodes.compose_layout import compose_layout_node
 from .nodes.fetch_sources import fetch_sources_node
+from .nodes.generate_thumbnail import generate_thumbnail_node
 from .nodes.publish import publish_node
 from .state import EditorialArticleState
 
@@ -23,6 +24,7 @@ def create_editorial_article_graph():
 
     builder.add_node("fetch_sources", fetch_sources_node)
     builder.add_node("compose_layout", compose_layout_node)
+    builder.add_node("generate_thumbnail", generate_thumbnail_node)
     builder.add_node("publish", publish_node)
 
     builder.add_edge(START, "fetch_sources")
@@ -34,8 +36,10 @@ def create_editorial_article_graph():
     builder.add_conditional_edges(
         "compose_layout",
         _route_after,
-        {"ok": "publish", "failed": END},
+        # generate_thumbnail 는 graceful — 실패해도 publish 로 진행
+        {"ok": "generate_thumbnail", "failed": END},
     )
+    builder.add_edge("generate_thumbnail", "publish")
     builder.add_edge("publish", END)
 
     return builder.compile()

packages/ai-server/src/editorial_article/models.py

@@ -15,6 +15,8 @@ class SectionSolution(BaseModel):
     brand: Optional[str] = None
     price: Optional[str] = None
     image_url: Optional[str] = None
+    original_url: Optional[str] = None     # 상품 페이지 (#429 commercial)
+    affiliate_url: Optional[str] = None    # 제휴 링크 (현재 미수집, 향후)
 
 
 class MagazineSection(BaseModel):
@@ -42,7 +44,8 @@ class MagazineLayout(BaseModel):
     schema_version: str = "1.0"
     title: str
     subtitle: Optional[str] = None
-    hero_image_url: Optional[str] = None
+    hero_image_url: Optional[str] = None     # 16:9 — nano-banana 결과 (#429)
+    thumbnail_url: Optional[str] = None      # 4:5 — nano-banana 결과 (#429)
     sections: list[MagazineSection]
 
 
@@ -58,6 +61,8 @@ class SourceSolution(BaseModel):
     price: Optional[str] = None
     image_url: Optional[str] = None
     sub_category: Optional[str] = None
+    original_url: Optional[str] = None     # 상품 페이지 (#429 commercial)
+    affiliate_url: Optional[str] = None
 
 
 class SourcePost(BaseModel):

packages/ai-server/src/editorial_article/nodes/fetch_sources.py

@@ -16,9 +16,12 @@
 
 
 async def _load_context(
-    db: DatabaseManager, recommendation_id: str
+    assets_db: DatabaseManager,
+    operation_db: DatabaseManager,
+    recommendation_id: str,
 ) -> RecommendationContext:
-    async with db.acquire() as conn:
+    # #429: recommendations 는 assets, posts/spots/solutions 는 operation.
+    async with assets_db.acquire() as conn:
         rec = await conn.fetchrow(
             """
             SELECT id, angle_title, angle_description, rationale,
@@ -28,14 +31,15 @@ async def _load_context(
             """,
             recommendation_id,
         )
-        if rec is None:
-            raise ValueError(f"recommendation {recommendation_id} not found")
+    if rec is None:
+        raise ValueError(f"recommendation {recommendation_id} not found")
 
-        post_ids = list(rec["source_post_ids"] or [])
-        sol_id_filter = set(str(x) for x in (rec["source_solution_ids"] or []))
+    post_ids = list(rec["source_post_ids"] or [])
+    sol_id_filter = set(str(x) for x in (rec["source_solution_ids"] or []))
 
-        posts: list[SourcePost] = []
-        if post_ids:
+    posts: list[SourcePost] = []
+    if post_ids:
+        async with operation_db.acquire() as conn:
             post_rows = await conn.fetch(
                 """
                 SELECT id, artist_name, group_name, context, image_url, style_tags
@@ -48,7 +52,8 @@ async def _load_context(
             sol_rows = await conn.fetch(
                 """
                 SELECT s.post_id,
-                       sol.id, sol.title, sol.thumbnail_url, sol.metadata
+                       sol.id, sol.title, sol.thumbnail_url, sol.metadata,
+                       sol.original_url, sol.affiliate_url
                   FROM public.spots s
                   JOIN public.solutions sol ON sol.spot_id = s.id
                  WHERE s.post_id = ANY($1::uuid[])
@@ -80,59 +85,64 @@ async def _load_context(
                             price=str(meta.get("price")) if meta.get("price") else None,
                             image_url=sr["thumbnail_url"],
                             sub_category=meta.get("sub_category"),
+                            original_url=sr["original_url"],
+                            affiliate_url=sr["affiliate_url"],
                         )
                     )
 
-            for pr in post_rows:
-                pid = str(pr["id"])
-                style_tags = pr["style_tags"] or []
-                if isinstance(style_tags, str):
-                    try:
-                        style_tags = json.loads(style_tags)
-                    except Exception:
-                        style_tags = []
-                posts.append(
-                    SourcePost(
-                        post_id=pid,
-                        artist_name=pr["artist_name"],
-                        group_name=pr["group_name"],
-                        context=pr["context"],
-                        image_url=pr["image_url"],
-                        style_tags=list(style_tags) if isinstance(style_tags, list) else [],
-                        solutions=sols_by_post.get(pid, []),
-                    )
+        for pr in post_rows:
+            pid = str(pr["id"])
+            style_tags = pr["style_tags"] or []
+            if isinstance(style_tags, str):
+                try:
+                    style_tags = json.loads(style_tags)
+                except Exception:
+                    style_tags = []
+            posts.append(
+                SourcePost(
+                    post_id=pid,
+                    artist_name=pr["artist_name"],
+                    group_name=pr["group_name"],
+                    context=pr["context"],
+                    image_url=pr["image_url"],
+                    style_tags=list(style_tags) if isinstance(style_tags, list) else [],
+                    solutions=sols_by_post.get(pid, []),
                 )
+            )
 
-        keywords = rec["keywords"] or []
-        if isinstance(keywords, str):
-            try:
-                keywords = json.loads(keywords)
-            except Exception:
-                keywords = []
-
-        return RecommendationContext(
-            recommendation_id=str(rec["id"]),
-            angle_title=rec["angle_title"],
-            angle_description=rec["angle_description"],
-            rationale=rec["rationale"],
-            keywords=list(keywords) if isinstance(keywords, list) else [],
-            source_posts=posts,
-        )
+    keywords = rec["keywords"] or []
+    if isinstance(keywords, str):
+        try:
+            keywords = json.loads(keywords)
+        except Exception:
+            keywords = []
+
+    return RecommendationContext(
+        recommendation_id=str(rec["id"]),
+        angle_title=rec["angle_title"],
+        angle_description=rec["angle_description"],
+        rationale=rec["rationale"],
+        keywords=list(keywords) if isinstance(keywords, list) else [],
+        source_posts=posts,
+    )
 
 
 async def fetch_sources_node(state: dict, config: RunnableConfig) -> dict:
-    """recommendation_id 로 posts + 솔루션 모두 로드."""
-    db: DatabaseManager | None = (config or {}).get("configurable", {}).get(
-        "operation_database_manager"
-    )
-    if db is None:
+    """recommendation (assets) + posts/spots/solutions (operation) 로드."""
+    cfg = (config or {}).get("configurable", {})
+    assets_db: DatabaseManager | None = cfg.get("assets_database_manager")
+    operation_db: DatabaseManager | None = cfg.get("operation_database_manager")
+    if assets_db is None or operation_db is None:
         return {
             "pipeline_status": "failed",
-            "error_log": ["fetch_sources: operation_database_manager missing"],
+            "error_log": [
+                "fetch_sources: db pools missing "
+                f"(assets={assets_db}, operation={operation_db})"
+            ],
         }
 
     try:
-        ctx = await _load_context(db, state["recommendation_id"])
+        ctx = await _load_context(assets_db, operation_db, state["recommendation_id"])
     except Exception as exc:
         logger.exception("fetch_sources failed")
         return {