[feat] 카드 카탈로그 DB 정본화 — config 결합 제거 + 학습 보존 자동반영

카드 카탈로그(negodata)가 Q-table action space 를 정의하는 정본이 되고, 카드 변경이
config 수정·학습 손실 없이 agent 에 자동 반영되는 고리를 완성.

- action space 정리: 카탈로그 전체(NGC-001~011, 11장) 고정, 견적별 선택은 축소가 아니라
  available_mask(_selection_mask) 로 처리 — action_id↔카드 대응을 견적마다 일정하게 유지해
  Q-table 학습 일관성 보장. 구 인덱스 방식(selected[action_id]) 폐기.
- ① 카탈로그 DB 정본화: action_mapping.type=db 면 registry 가 card.nego_cards(user_id NULL,
  number 순) 조회로 action_to_card 동적 구성(파일은 폴백). port/adapter(card_catalog_*).
  _base=type:db. → negodata 카드 추가/삭제 시 config 수정 불필요.
- ② 차원 변경 학습 보존 마이그레이션: migrate_active_version_dim — 겹치는 셀 복사
  (append/truncate 안전) + 새 카드 fresh. model_store.load 가 차원 불일치 시 호출.
- ③ reload 엔드포인트: /v1/catalog-refresh(테넌트) · /v1/catalog-refresh-all(전역, 화이트리스트).
- ④ 브랜드: company_profile_repo — 자동 온보딩 고객사(company_id UUID)는
  company.companies.name 으로 {company_name} 채움. 데모 테넌트는 파일 유지.
- 크로스서비스: negodata card_service 가 공용 nego 카드 변경 시 agent_notify 로 전역 리로드 알림
  (best-effort, is_test skip). config 에 agent_base_url.
- 하니스 episodes 400→600(action 11 수렴). 테스트 갱신·추가로 agent 98/98.

알려진 갭(후속): per-company 카탈로그 스코프(회사 카드도 action space 포함), 카탈로그 중간
삭제 시 카드번호 기반 마이그레이션.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

agent/eval_harness/configs/exp_default.yaml

@@ -1,5 +1,5 @@
 # 알고리즘 비교 실험 기본 설정 (H5). E2E: python -m eval_harness.runner --config configs/exp_default.yaml --tenant ktcommerce
-episodes: 400          # 정책당 협상 에피소드 수
+episodes: 600          # 정책당 협상 에피소드 수 (action 11장 탐색 수렴 위해 상향)
 seed: 42               # 재현용 (구매자 randomness 페어드)
 max_turns: 5           # 협상 라운드 상한
 n_good_cards: 3        # 시뮬 구매자: 효과 좋은 카드 수

agent/negotiation/cards/adapters/card_catalog_db.py

@@ -0,0 +1,39 @@
+"""CardCatalogDbRepository — card.nego_cards 에서 카탈로그(action space) read-only 조회.
+
+정렬: number 오름차순(zero-padded NGC-001..NGC-011 이라 문자열 정렬로 안정적) → action_id.
+스코프: o2o 기본 제공 카드(user_id IS NULL). 회사 전용 카탈로그는 후속(스코프 확장).
+스키마 소유권: card 스키마는 backend/negodata 소유 — read-only. 경량 table()/column() 구성.
+"""
+
+from typing import List, Tuple
+
+from sqlalchemy import asc, column, select, table
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.enums import ErrorType
+from common.logger import LOG
+from negotiation.cards.ports.card_catalog_port import ICardCatalogRepository
+
+_NEGO_CARDS = table(
+    "nego_cards",
+    column("number"), column("user_id"), column("deleted"),
+    schema="card",
+)
+
+
+class CardCatalogDbRepository(ICardCatalogRepository):
+    async def get_nego_catalog(self, cdb: AsyncSession) -> Tuple[ErrorType, List[str]]:
+        try:
+            query = (
+                select(_NEGO_CARDS.c.number)
+                .where(_NEGO_CARDS.c.user_id.is_(None), _NEGO_CARDS.c.deleted == False)  # noqa: E712
+                .order_by(asc(_NEGO_CARDS.c.number))
+            )
+            err_type, rows = await DB_SESSION_MNG.execute(cdb, query, "get_nego_catalog failed.", raise_error=False)
+            if err_type != ErrorType.SUCCESS or not rows:
+                return err_type, []
+            return ErrorType.SUCCESS, [str(r) for r in rows if r]
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED, []

agent/negotiation/cards/ports/card_catalog_port.py

@@ -0,0 +1,24 @@
+"""ICardCatalogRepository — 협상카드 카탈로그(action space 정의) DB 조회 포트.
+
+카탈로그 = Q-table 의 action 축을 정의하는 "카드 목록". negodata 가 관리하는 card.nego_cards 가
+정본이며, agent 는 이를 조회해 action_id(0..N-1) ↔ 카드번호 매핑을 구성한다. tenant.yaml 의
+action_to_card 하드코딩을 대체한다(config 결합 제거 — 카탈로그는 DB, config 는 튜닝만).
+
+주의: 카탈로그 크기 = Q-table action 차원. 카드 수가 바뀌면 재차원화(버전 이벤트)가 필요하므로
+카탈로그 변경은 드문 이벤트여야 한다.
+"""
+
+from abc import ABC, abstractmethod
+from typing import List, Tuple
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from common.enums import ErrorType
+
+
+class ICardCatalogRepository(ABC):
+    @abstractmethod
+    async def get_nego_catalog(self, cdb: AsyncSession) -> Tuple[ErrorType, List[str]]:
+        """action space 를 정의하는 일반 협상카드 번호 목록(정렬됨). action_id = 리스트 인덱스.
+        비었으면([]) 호출부가 파일 action_to_card 로 폴백한다."""
+        ...

agent/negotiation/policy/model_store.py

@@ -10,6 +10,7 @@ P5/H 트랙에서 transition_id 기반으로 정교화한다.
 import math
 from typing import Tuple
 
+from common.logger import LOG
 from negotiation.policies.qtable_policy import UCBQTablePolicy
 from negotiation.qtable.domain.model.q_table import QTable
 from negotiation.qtable.infra.repository.learning_repository import LearningRepository
@@ -31,6 +32,17 @@ class QTablePolicyStore:
         #  ③ 차원 불일치/base 없음 → 휴리스틱 빈 버전
         err, active = await repo.read(lambda s: repo.get_active_version(s))
         if active is not None:
+            if active.action_space_size != A or active.state_space_size != S:
+                # 카탈로그 카드 수(또는 상태 차원) 변경 → 학습 보존 마이그레이션.
+                #   겹치는 셀 복사(append/truncate 안전) + 새 카드 fresh. 실패 시 기존 버전 유지(load 가 reshape).
+                migrated = await repo.migrate_active_version_dim(
+                    old_version=active, state_space_size=S, action_space_size=A,
+                    learning_rate=lr, discount_factor=gamma, version_name=f"v_migrated_a{A}")
+                if migrated is not None:
+                    LOG.i(f"[QTablePolicyStore] 차원 변경 마이그레이션 company={engine.company_id} "
+                          f"{active.action_space_size}→{A} action (학습 보존)")
+                version_id = migrated or active.version_id
+            else:
                 version_id = active.version_id
         else:
             version_id = None

agent/negotiation/qtable/infra/repository/learning_repository.py

@@ -227,6 +227,54 @@ class LearningRepository:
         err = await DB_SESSION_MNG.execute_lambda_run([DBType.MAIN.value], [_create])
         return vid if err == ErrorType.SUCCESS else None
 
+    async def migrate_active_version_dim(self, *, old_version, state_space_size: int, action_space_size: int,
+                                         learning_rate: float, discount_factor: float,
+                                         version_name: str = "v_migrated") -> Optional[uuid.UUID]:
+        """활성 Q-table 을 새 차원으로 마이그레이션 (카탈로그 카드 수 변경 = action 차원 변경 시).
+
+        겹치는 (state < S, action < A) 셀만 복사한다 → **카탈로그 끝에 카드 추가(append)/삭제(truncate)에
+        안전**(action_id↔카드 위치 불변). 늘어난 새 action(새 카드)은 fresh(0, UCB 가 우선 탐험).
+        기존 활성 버전은 비활성화하고 새 버전을 활성화한다. 실패 시 None(호출자가 기존 버전 유지).
+
+        ⚠️ 한계: 카탈로그 **중간 삽입·삭제**는 action_id 가 밀려 학습이 어긋날 수 있다 — 카드번호 기반
+        매핑(버전별 카탈로그 스냅샷 저장, tenant_action_cards 활용)이 후속 과제다.
+        """
+        qcells, vcells = await self.load_cells(old_version.version_id)
+        vid = uuid.uuid4()
+        S, A = state_space_size, action_space_size
+
+        async def _create(s: AsyncSession) -> ErrorType:
+            # 기존 활성 비활성화 → 새 버전 활성 삽입 (동시 2개 활성 방지).
+            e = await DB_SESSION_MNG.add(s, update(QTableVersion).where(
+                QTableVersion.company_id == self.company_id,
+                QTableVersion.version_id == old_version.version_id).values(is_active=False))
+            if e != ErrorType.SUCCESS:
+                return e
+            ver = QTableVersion(
+                version_id=vid, company_id=self.company_id, version_name=version_name, scope=2,
+                base_version_id=old_version.version_id, state_space_size=S, action_space_size=A,
+                learning_rate=learning_rate, discount_factor=discount_factor, is_active=True,
+            )
+            e = await DB_SESSION_MNG.insert(s, ver)
+            if e != ErrorType.SUCCESS:
+                return e
+            qobjs = [QValue(company_id=self.company_id, version_id=vid, state_index=st, action_id=a, q_value=q)
+                     for st, a, q in qcells if st < S and a < A]
+            vobjs = [VisitCount(company_id=self.company_id, version_id=vid, state_index=st, action_id=a, count=c)
+                     for st, a, c in vcells if st < S and a < A]
+            if qobjs:
+                e = await DB_SESSION_MNG.insert(s, qobjs)
+                if e != ErrorType.SUCCESS:
+                    return e
+            if vobjs:
+                e = await DB_SESSION_MNG.insert(s, vobjs)
+                if e != ErrorType.SUCCESS:
+                    return e
+            return ErrorType.SUCCESS
+
+        err = await DB_SESSION_MNG.execute_lambda_run([DBType.MAIN.value], [_create])
+        return vid if err == ErrorType.SUCCESS else None
+
     async def load_cells(self, version_id) -> Tuple[List[tuple], List[tuple]]:
         """(q_cells, visit_cells) — q_cells: (state,action,q), visit_cells: (state,action,count). 자사 스코프."""
         def _q(s):

agent/router/middleware/tenant_middleware.py

@@ -20,9 +20,8 @@ from config.server_configs import agent_config
 
 _TENANT_HEADER = "X-Tenant-ID"
 
-# 테넌트 식별이 필요 없는 경로 (헬스/문서/스키마/데모 UI).
+# 테넌트 식별이 필요 없는 경로 (헬스/문서/스키마/데모 UI + 전역 카탈로그 리로드=내부 운영).
-_WHITELIST_PREFIXES = ("/healthz", "/health", "/v1/health", "/docs", "/redoc", "/openapi.json", "/demo")
+_WHITELIST_PREFIXES = ("/healthz", "/health", "/v1/health", "/docs", "/redoc", "/openapi.json", "/demo", "/v1/catalog-refresh-all")
-
 
 class TenantMiddleware(BaseHTTPMiddleware):
     async def dispatch(self, request: Request, call_next):

agent/router/v1/card/card.py

@@ -6,13 +6,13 @@ PoC: action_id ↔ card_id 매핑을 learning.tenant_action_cards 에 둔다(con
 
 from typing import Optional
 
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, Request
 from pydantic import BaseModel
 
-from common.enums import ErrorType
+from common.enums import EXCEPTION_TENANT_HEADER_MISSING, ErrorType
 from negotiation.qtable.infra.repository.learning_repository import LearningRepository
 from router.deps import get_tenant_engine
-from tenancy.registry import TenantEngine
+from tenancy.registry import TenantEngine, tenant_registry
 
 router = APIRouter(prefix="/v1", tags=["Card"], responses={404: {"description": "Not found"}})
 
@@ -41,6 +41,33 @@ async def card_update(req: CardUpdateReq, engine: TenantEngine = Depends(get_ten
             "action_id": req.action_id, "card_id": req.card_id, "desc": err.name}
 
 
+@router.post("/catalog-refresh-all", summary="공용 카탈로그 변경 전역 반영(모든 엔진 캐시 클리어)")
+async def catalog_refresh_all():
+    """공용 카드(card.nego_cards user_id NULL)는 모든 테넌트의 action space 를 정의하므로,
+    변경 시 전역 반영이 필요하다. 캐시된 엔진을 전부 비워 다음 요청에서 최신 카탈로그로 재조립한다.
+    o2o 운영/negodata 공용카드 발행 훅에서 호출. 테넌트 헤더 불필요(미들웨어 화이트리스트) — 내부망 전용.
+    """
+    cleared = tenant_registry.clear_all()
+    return {"success": True, "cleared_engines": cleared}
+
+
+@router.post("/catalog-refresh", summary="카드 카탈로그 변경 반영(엔진 재조립)")
+async def catalog_refresh(request: Request):
+    """negodata 가 카드 카탈로그(card.nego_cards)를 추가/삭제(발행)한 뒤 호출한다.
+
+    해당 테넌트의 캐시된 엔진을 재조립해 → ① DB 카탈로그로 action space 재구성(type:db)
+    → 다음 협상 로드 시 ② 차원 변경이면 Q-table 학습 보존 마이그레이션이 걸린다.
+    엔진은 캐시되므로 이 호출 없이는 카탈로그 변경이 반영되지 않는다.
+    """
+    tenant_id = getattr(request.state, "tenant_id", None)
+    if not tenant_id:
+        raise EXCEPTION_TENANT_HEADER_MISSING
+    engine = await tenant_registry.reload(tenant_id)
+    if engine is None:
+        return {"success": False, "desc": "unregistered tenant or reload failed", "tenant_id": tenant_id}
+    return {"success": True, "company_id": engine.company_id, "action_space_size": engine.action_space_size}
+
+
 @router.get("/card-search", summary="카드 검색(전체 매핑 또는 card_id 조회)")
 async def card_search(card_id: Optional[str] = None, engine: TenantEngine = Depends(get_tenant_engine)):
     mapping = await _merged_mapping(engine)

agent/services/chat_service.py

@@ -7,6 +7,8 @@ ChatEngine(동기 step 전이) + UCB Q-Table(가격협상 카드선택·학습)
 import uuid
 from typing import Optional
 
+import numpy as np
+
 from common.enums import DBType, ErrorType
 from common.database.db_session_manager import DB_SESSION_MNG
 from common.logger import LOG
@@ -188,7 +190,11 @@ class ChatService:
             LOG.e_no_callstack(f"[ChatService] state error: {ex}")
             return
         policy, version_id, repo = await QTablePolicyStore.load(engine)
-        ctx = PolicyContext(state_index=idx, snapshot=snap, action_space_size=session.action_space_size,
+        # action space 는 카탈로그 전체(engine.action_space_size)로 고정 — action_id↔카드 대응을
+        # 견적마다 일정하게 유지해 Q-table 학습 일관성을 지킨다. 견적 선택은 축소가 아니라
+        # available_mask 로 처리한다(선택 카드만 pickable, 사용분 제외).
+        ctx = PolicyContext(state_index=idx, snapshot=snap, action_space_size=engine.action_space_size,
+                            available_mask=self._selection_mask(engine, session),
                             episode=EpisodeState(used_action_ids=set(session.used_action_ids)))
         decision = policy.select(ctx)
         session.used_action_ids.add(decision.action_id)
@@ -244,11 +250,32 @@ class ChatService:
 
     @staticmethod
     def _card_id_for_action(engine: TenantEngine, session: ChatSession, action_id: int) -> Optional[str]:
-        selected = session.context.get("selected_nego_card_numbers") or []
+        # action_id ↔ 카드는 테넌트 매핑(action_to_card)으로 고정한다. 견적 선택은 available_mask 로
-        if 0 <= action_id < len(selected):
+        # 걸러지므로 여기서 selected 리스트를 인덱싱하지 않는다 — 인덱싱하면 견적마다 action_id 의미가
-            return str(selected[action_id])
+        # 달라져(같은 action_id 가 다른 카드) Q-table 학습이 오염된다.
         return engine.mapper.get_card_id(action_id)
 
+    @staticmethod
+    def _selection_mask(engine: TenantEngine, session: ChatSession) -> Optional[np.ndarray]:
+        """견적에서 선택한 카드(selected_nego_card_numbers)만 pickable 로 하는 available_mask.
+
+        action space 전체(engine.action_space_size) 크기의 bool 배열. 선택 카드의 action_id 만 True,
+        이미 사용한 action 은 False. 선택이 없거나(직접호출/데모) 매핑 불가면 None → 전체 허용(폴백).
+        번호(card.nego_cards.number)와 action_to_card 값이 일치해야 매핑된다.
+        """
+        selected = session.context.get("selected_nego_card_numbers") or []
+        if not selected:
+            return None
+        used = set(session.used_action_ids)
+        selected_ids = {engine.mapper.get_action_id(str(n)) for n in selected}
+        selected_ids.discard(None)
+        if not selected_ids:
+            return None  # 매핑에 없는 번호뿐 → 폴백(전체 허용)
+        return np.array(
+            [(a in selected_ids and a not in used) for a in range(engine.action_space_size)],
+            dtype=bool,
+        )
+
     async def _log(self, repo: LearningRepository, session, state_index, action_id, card_id, snap, reward, propensity, done):
         data = {
             "session_id": session.session_id, "state_index": state_index, "action_id": action_id,

agent/tenancy/company_profile_repo.py

@@ -0,0 +1,64 @@
+"""회사 프로필(브랜드) DB 조회 — 자동 온보딩 고객사의 {company_name} 을 DB 에서 채운다.
+
+스크립트의 {company_name} 은 지금까지 tenant.yaml resources.company_name(_base 기본값)에서 왔다.
+자동 온보딩 고객사(전용 yaml 없음)는 전부 _base 브랜드로 나가므로, 실제 회사명(company.companies.name)을
+company_id 로 조회해 엔진 조립 시 덮어쓴다. 회사 정보라 튜닝 오버레이와는 별개 관심사다.
+
+스키마 소유권: company 스키마는 backend/negodata 소유 — read-only. 경량 table()/column() 구성.
+"""
+
+import uuid
+from abc import ABC, abstractmethod
+from typing import Optional, Tuple
+
+from sqlalchemy import column, select, table
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.enums import DBType, DBWRType, ErrorType
+from common.logger import LOG
+
+_COMPANIES = table("companies", column("company_id"), column("name"), column("deleted"), schema="company")
+
+
+class ICompanyProfileRepository(ABC):
+    @abstractmethod
+    async def get_company_name(self, cdb: AsyncSession, company_id) -> Tuple[ErrorType, Optional[str]]:
+        """company.companies.name by company_id. 없으면 None."""
+        ...
+
+
+class CompanyProfileRepository(ICompanyProfileRepository):
+    async def get_company_name(self, cdb: AsyncSession, company_id) -> Tuple[ErrorType, Optional[str]]:
+        try:
+            query = (
+                select(_COMPANIES.c.name)
+                .where(_COMPANIES.c.company_id == company_id, _COMPANIES.c.deleted == False)  # noqa: E712
+                .limit(1)
+            )
+            err_type, rows = await DB_SESSION_MNG.execute(cdb, query, "get_company_name failed.", raise_error=False)
+            if err_type != ErrorType.SUCCESS or not rows or not rows[0]:
+                return err_type, None
+            return ErrorType.SUCCESS, str(rows[0])
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED, None
+
+
+async def resolve_company_name(repo: ICompanyProfileRepository, tenant_key: str) -> Optional[str]:
+    """tenant_key(= company_id UUID)로 회사명 조회. 비-UUID(데모 테넌트)면 조회 안 함(None).
+    세션/트랜잭션 경계를 여기서 관리(execute_lambda)한다."""
+    try:
+        cid = uuid.UUID(tenant_key)
+    except (ValueError, TypeError):
+        return None  # ktcommerce 등 데모 테넌트명 → 파일 브랜드 유지
+
+    async def _q(s):
+        _, name = await repo.get_company_name(s, cid)
+        return name
+
+    try:
+        return await DB_SESSION_MNG.execute_lambda(DBType.MAIN.value, DBWRType.DB_READ.value, _q)
+    except Exception as ex:
+        LOG.e_no_callstack(f"[company_profile] 회사명 조회 실패 company_id={tenant_key}: {ex}")
+        return None

agent/tenancy/registry.py

@@ -11,11 +11,22 @@ import asyncio
 from collections import defaultdict
 from typing import Dict, Optional
 
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.enums import DBType, DBWRType, ErrorType
 from common.logger import LOG
 from negotiation.cards.action_card_mapper import ActionCardMapper
+from negotiation.cards.adapters.card_catalog_db import CardCatalogDbRepository
+from negotiation.cards.ports.card_catalog_port import ICardCatalogRepository
+from tenancy.company_profile_repo import (
+    CompanyProfileRepository,
+    ICompanyProfileRepository,
+    resolve_company_name,
+)
 from tenancy.config import TenantConfig
 from tenancy.config_loader import TenantConfigLoader
 
+_CARD_SOURCE_DB = "db"
+
 
 class TenantEngine:
     """한 테넌트의 협상 엔진 조립체 (불변 협력자 보관).
@@ -49,9 +60,13 @@ class EngineFactory:
 
 
 class TenantEngineRegistry:
-    def __init__(self, loader: Optional[TenantConfigLoader] = None, factory: type[EngineFactory] = EngineFactory):
+    def __init__(self, loader: Optional[TenantConfigLoader] = None, factory: type[EngineFactory] = EngineFactory,
+                 catalog_repo: Optional[ICardCatalogRepository] = None,
+                 company_repo: Optional[ICompanyProfileRepository] = None):
         self._loader = loader or TenantConfigLoader()
         self._factory = factory
+        self._catalog_repo = catalog_repo or CardCatalogDbRepository()
+        self._company_repo = company_repo or CompanyProfileRepository()
         self._engines: Dict[str, TenantEngine] = {}
         self._locks: Dict[str, asyncio.Lock] = defaultdict(asyncio.Lock)
 
@@ -69,8 +84,7 @@ class TenantEngineRegistry:
                 return cached
             if not self.is_registered(tenant_id):
                 raise KeyError(f"unregistered tenant: {tenant_id}")
-            config = self._loader.load(tenant_id)
+            engine = await self._build(tenant_id)
-            engine = self._factory.build(config)
             self._engines[tenant_id] = engine
             LOG.i(f"[TenantEngineRegistry] built engine tenant_id={tenant_id} "
                   f"state={engine.state_space_size} action={engine.action_space_size}")
@@ -83,11 +97,52 @@ class TenantEngineRegistry:
             self._engines.pop(tenant_id, None)
             if not self.is_registered(tenant_id):
                 return None
-            config = self._loader.load(tenant_id)
+            engine = await self._build(tenant_id)
-            engine = self._factory.build(config)
             self._engines[tenant_id] = engine
             return engine
 
+    # ---- 조립 (카탈로그 DB 유래 반영) ----------------------------------
+    async def _build(self, tenant_id: str) -> TenantEngine:
+        config = self._loader.load(tenant_id)
+        # action_mapping.type == "db" 면 카드 카탈로그(card.nego_cards)에서 action_to_card 를 동적 구성.
+        # Q-table action 축을 config 파일이 아니라 negodata 카탈로그가 정의한다(결합 제거).
+        if config.action_mapping.type == _CARD_SOURCE_DB:
+            await self._apply_db_catalog(config)
+        # 브랜드: 자동 온보딩 고객사(company_id=UUID)면 company.companies.name 으로 {company_name} 을 덮어쓴다.
+        #   데모 테넌트(비-UUID)·미조회 시 파일 resources.company_name 유지.
+        await self._apply_company_brand(config, tenant_id)
+        return self._factory.build(config)
+
+    async def _apply_company_brand(self, config: TenantConfig, tenant_id: str) -> None:
+        name = await resolve_company_name(self._company_repo, config.company_id or tenant_id)
+        if name:
+            config.resources.company_name = name
+
+    async def _apply_db_catalog(self, config: TenantConfig) -> None:
+        """DB 카탈로그로 config.action_mapping.action_to_card 를 덮어쓴다(성공 시).
+        비었거나 DB 불가면 config 의 파일 action_to_card 를 그대로 폴백 사용한다."""
+        async def _q(s):
+            _, numbers = await self._catalog_repo.get_nego_catalog(s)
+            return numbers
+
+        try:
+            numbers = await DB_SESSION_MNG.execute_lambda(DBType.MAIN.value, DBWRType.DB_READ.value, _q)
+        except Exception as ex:
+            LOG.e_no_callstack(f"[TenantEngineRegistry] 카탈로그 조회 실패 tenant={config.tenant_id}: {ex} → 파일 폴백")
+            return
+        if not numbers:
+            LOG.w(f"[TenantEngineRegistry] 카탈로그 비어있음 tenant={config.tenant_id} → 파일 action_to_card 폴백")
+            return
+        config.action_mapping.action_to_card = {str(i): num for i, num in enumerate(numbers)}
+
+    def clear_all(self) -> int:
+        """캐시된 엔진 전체를 비운다(공용 카탈로그 변경 등 전역 반영용). 반환: 비운 엔진 수.
+        다음 요청에서 각 테넌트 엔진이 최신 카탈로그/config 로 재조립된다."""
+        n = len(self._engines)
+        self._engines.clear()
+        self._loader.invalidate()   # 인자 없이 = 전체 config 캐시 무효화
+        return n
+
     def cached_tenants(self) -> list[str]:
         return list(self._engines.keys())
 

agent/tenants/_base/tenant.yaml

@@ -49,9 +49,14 @@ policy:
     epsilon: 1.0e-6
 
 action_mapping:
-  type: file
+  # db: action_to_card 를 card.nego_cards 카탈로그(user_id NULL, number 순)에서 동적 구성 — 정본.
-  # 기본 9카드 — 자동 온보딩(신규 company_id) 테넌트가 물려받는 카드 공간(base 정책 162×9와 정합).
+  #     카드가 negodata 에서 추가/삭제되면 action space 가 자동 반영된다(config 수정 불필요).
-  # 테넌트는 자사 카탈로그(card.nego_cards/tenant_action_cards)로 override 한다.
+  #     아래 action_to_card 는 DB 카탈로그가 비어있을 때만 쓰는 폴백(정합용 스냅샷)이다.
+  # file: 아래 action_to_card 를 그대로 사용(데모/오프라인).
+  type: db
+  # [폴백] 카탈로그 11장(카드_기획문서 일반카드) 스냅샷 — action_id 0~10 ↔ NGC-001~NGC-011.
+  #   Q-table action 차원 = 카탈로그 크기. 견적별 선택은 action space 축소가 아니라 available_mask 로 처리
+  #   (선택 카드만 pickable) — action_id↔카드 대응을 견적마다 일정하게 유지해 학습 일관성 보장.
   action_to_card:
     "0": "NGC-001"
     "1": "NGC-002"
@@ -62,6 +67,8 @@ action_mapping:
     "6": "NGC-007"
     "7": "NGC-008"
     "8": "NGC-009"
+    "9": "NGC-010"
+    "10": "NGC-011"
 
 cards:
   # file: scripts_cards.json(파일) 사용. backoffice_db: card.nego_cards.script(negodata 편집 정본)를

agent/tenants/imarketkorea/tenant.yaml

@@ -23,7 +23,7 @@ reward:
 
 action_mapping:
   type: file
-  action_to_card:            # 동일 차원(9) 유지 → warm-start 가능. 합성 데모 코드(테넌트 B).
+  action_to_card:            # 동일 차원(11) 유지 → warm-start 가능. 합성 데모 코드(테넌트 B).
     "0": "NGC-B001"
     "1": "NGC-B002"
     "2": "NGC-B003"
@@ -33,6 +33,8 @@ action_mapping:
     "6": "NGC-B007"
     "7": "NGC-B008"
     "8": "NGC-B009"
+    "9": "NGC-B010"
+    "10": "NGC-B011"
 
 llm:
   enabled: false

agent/tenants/ktcommerce/tenant.yaml

@@ -8,7 +8,7 @@ company_id: null            # P3 시드 시 company.companies.company_id(uuid)
 
 action_mapping:
   type: file
-  action_to_card:           # 우리 중립 데모 카드 코드(합성). 실제 카드 코드 아님.
+  action_to_card:           # 우리 중립 데모 카드 코드(합성). 실제 카드 코드 아님. 카탈로그 11장(_base 와 정합).
     "0": "NGC-A001"
     "1": "NGC-A002"
     "2": "NGC-A003"
@@ -18,6 +18,8 @@ action_mapping:
     "6": "NGC-A007"
     "7": "NGC-A008"
     "8": "NGC-A009"
+    "9": "NGC-A010"
+    "10": "NGC-A011"
 
 llm:
   enabled: false            # P7 에서 테넌트별 자격증명 주입

agent/tests/test_agent_integration.py

@@ -53,7 +53,7 @@ async def test_4_4_company_id_auto_onboard():
     eng = await _reg().get_engine(COMPANY_ID)
     assert eng.tenant_id == COMPANY_ID
     assert eng.company_id == COMPANY_ID            # 학습/세션이 이 company_id 로 격리
-    assert eng.action_space_size == 9              # base 기본 카드(162×9 정합)
+    assert eng.action_space_size == 11              # base 기본 카드(162×11 정합)
     assert eng.state_space_size == 162
 
 

agent/tests/test_context_loader.py

@@ -204,6 +204,9 @@ async def test_loader_with_crud_double(db_engine):
         async def count_item_session_suppliers(self, cdb, item_id):
             return ErrorType.SUCCESS, 0  # 이력도 없음 → NONE
 
+        async def get_quotation_card_numbers(self, cdb, quotation_id):
+            return ErrorType.SUCCESS, (["NGC-003", "NGC-008"], ["WC-02"])  # 견적 선택 카드
+
     ctx = await NegotiationContextLoader(crud=_FakeCRUD()).load(str(uuid.uuid4()))
     assert ctx is not None
     assert ctx.rq_type == "재견적"            # qt_type=2(1:N)
@@ -213,6 +216,8 @@ async def test_loader_with_crud_double(db_engine):
     assert ctx.revenue_amount == 12_000_000.0
     assert ctx.distribution_code == "B"      # supply_type=3(총판) → B
     assert ctx.partner_type is PartnerType.NONE
+    assert ctx.selected_nego_card_numbers == ["NGC-003", "NGC-008"]
+    assert ctx.selected_wild_card_numbers == ["WC-02"]
 
 
 @pytest.mark.asyncio

agent/tests/test_p1_tenant_config.py

@@ -49,7 +49,7 @@ def test_platform_neutral_defaults_load():
 def test_state_space_and_action_space_size():
     cfg = _loader().load("ktcommerce")
     assert cfg.state.state_space_size == 162  # 3×3×3×3×2 (차원 구성은 기능적 설계)
-    assert cfg.action_mapping.action_space_size == 9
+    assert cfg.action_mapping.action_space_size == 11
     # 합성 데모 카드 코드 (우리 스킴)
     assert cfg.action_mapping.action_to_card["0"] == "NGC-A001"
     assert cfg.action_mapping.action_to_card["8"] == "NGC-A009"
@@ -77,15 +77,15 @@ def test_second_tenant_overrides_merged_on_base():
     assert cfg.policy.type == "ucb"
     # 차원은 데모 테넌트 A 와 동일(162) → base warm-start 호환(P5)
     assert cfg.state.state_space_size == 162
-    assert cfg.action_mapping.action_space_size == 9
+    assert cfg.action_mapping.action_space_size == 11
     assert cfg.action_mapping.action_to_card["0"] == "NGC-B001"
 
 
 def test_base_self_does_not_inherit():
     cfg = _loader().load("_base")
     assert cfg.tenant_id == "_base"
-    # _base 는 기본 9카드(자동 온보딩 테넌트가 물려받음, base 정책 162×9 정합)
+    # _base 는 기본 11카드(자동 온보딩 테넌트가 물려받음, base 정책 162×11 정합)
-    assert cfg.action_mapping.action_space_size == 9
+    assert cfg.action_mapping.action_space_size == 11
 
 
 def test_is_registered():

agent/tests/test_p2_state_reward_mapper.py

@@ -129,7 +129,7 @@ def test_reward_deterministic_and_config_driven():
 def test_action_card_mapper_roundtrip_and_mask():
     cfg = _cfg("ktcommerce")
     mapper = ActionCardMapper(cfg.action_mapping)
-    assert mapper.action_space_size == 9
+    assert mapper.action_space_size == 11
     assert mapper.get_card_id(0) == "NGC-A001"
     assert mapper.get_action_id("NGC-A001") == 0
     assert mapper.get_card_id(99) is None
@@ -139,7 +139,7 @@ def test_action_card_mapper_roundtrip_and_mask():
     assert mask.dtype == np.bool_
     assert mask[0] == False and mask[3] == False
     assert mask[1] == True
-    assert mask.sum() == 7
+    assert mask.sum() == 9
 
     # reload 로 다른 테넌트 카드셋 교체
     mapper.reload(_cfg("imarketkorea").action_mapping)

agent/tests/test_p4_registry_middleware.py

@@ -35,7 +35,7 @@ async def test_two_tenants_distinct_engines():
     assert e1.mapper.get_card_id(0) == "NGC-A001"
     assert e2.mapper.get_card_id(0) == "NGC-B001"
     # 차원
-    assert e1.state_space_size == 162 and e1.action_space_size == 9
+    assert e1.state_space_size == 162 and e1.action_space_size == 11
 
 
 @pytest.mark.asyncio
@@ -69,7 +69,7 @@ async def test_unregistered_company_id_auto_onboards():
     reg = _registry()
     # 미등록 company_id(uuid)는 _base 자동 온보딩 → 엔진 생성됨(베이스 9카드, 162 state).
     eng = await reg.get_engine("00000000-0000-0000-0000-000000000001")
-    assert eng.action_space_size == 9 and eng.state_space_size == 162
+    assert eng.action_space_size == 11 and eng.state_space_size == 162
     assert eng.company_id == "00000000-0000-0000-0000-000000000001"
     assert reg.is_registered("ktcommerce") is True
     # 빈 키만 미등록 → KeyError
@@ -97,6 +97,84 @@ def test_episode_state_is_request_scoped():
     assert not hasattr(TenantEngine, "used_action_ids")
 
 
+@pytest.mark.asyncio
+async def test_action_space_from_db_catalog(db_engine):
+    """action_mapping.type=db(_base) 면 카드 카탈로그(DB)가 action space 를 정의한다 — config 파일이 아님.
+    negodata 에서 카드가 추가/삭제되면 config 수정 없이 action space 가 반영됨을 의미."""
+    from common.enums import ErrorType
+    from negotiation.cards.ports.card_catalog_port import ICardCatalogRepository
+
+    class _FakeCatalog(ICardCatalogRepository):
+        async def get_nego_catalog(self, cdb):
+            return ErrorType.SUCCESS, ["NGC-001", "NGC-002", "NGC-003"]  # 3장짜리 카탈로그(파일은 11장)
+
+    reg = TenantEngineRegistry(
+        loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0),
+        catalog_repo=_FakeCatalog(),
+    )
+    eng = await reg.get_engine("_base")   # _base = type:db
+    assert eng.action_space_size == 3      # DB 카탈로그(3)가 정의 — 파일 폴백(11) 아님
+    assert eng.mapper.get_card_id(0) == "NGC-001"
+    assert eng.mapper.get_card_id(2) == "NGC-003"
+    assert eng.mapper.get_action_id("NGC-002") == 1
+
+
+@pytest.mark.asyncio
+async def test_action_space_falls_back_to_file_when_catalog_empty(db_engine):
+    """카탈로그가 비면(신규/미시드 DB) 파일 action_to_card 로 폴백한다."""
+    from common.enums import ErrorType
+    from negotiation.cards.ports.card_catalog_port import ICardCatalogRepository
+
+    class _EmptyCatalog(ICardCatalogRepository):
+        async def get_nego_catalog(self, cdb):
+            return ErrorType.SUCCESS, []
+
+    reg = TenantEngineRegistry(
+        loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0),
+        catalog_repo=_EmptyCatalog(),
+    )
+    eng = await reg.get_engine("_base")
+    assert eng.action_space_size == 11     # 파일 폴백(11장 스냅샷)
+    assert eng.mapper.get_card_id(0) == "NGC-001"
+
+
+@pytest.mark.asyncio
+async def test_company_brand_from_db_for_auto_onboard(db_engine):
+    """자동 온보딩 고객사(company_id=UUID)는 company.companies.name 으로 {company_name} 을 채운다."""
+    import uuid as _uuid
+    from common.enums import ErrorType
+    from tenancy.company_profile_repo import ICompanyProfileRepository
+
+    class _FakeCompany(ICompanyProfileRepository):
+        async def get_company_name(self, cdb, company_id):
+            return ErrorType.SUCCESS, "풀무원"
+
+    reg = TenantEngineRegistry(
+        loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0),
+        company_repo=_FakeCompany(),
+    )
+    eng = await reg.get_engine(str(_uuid.uuid4()))   # UUID → 자동 온보딩 + 브랜드 DB
+    assert eng.config.resources.company_name == "풀무원"
+
+
+@pytest.mark.asyncio
+async def test_demo_tenant_keeps_file_brand(db_engine):
+    """데모 테넌트(비-UUID)는 회사명 조회 없이 파일 브랜드 유지."""
+    from common.enums import ErrorType
+    from tenancy.company_profile_repo import ICompanyProfileRepository
+
+    class _FakeCompany(ICompanyProfileRepository):
+        async def get_company_name(self, cdb, company_id):
+            return ErrorType.SUCCESS, "USED-ONLY-IF-QUERIED"
+
+    reg = TenantEngineRegistry(
+        loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0),
+        company_repo=_FakeCompany(),
+    )
+    eng = await reg.get_engine("ktcommerce")   # 비-UUID → 조회 안 함
+    assert eng.config.resources.company_name == "데모상사 A"
+
+
 @pytest.mark.asyncio
 async def test_middleware_header_missing_unregistered_registered(client):
     # 헤더 누락 → 400

agent/tests/test_p5_warmstart.py

@@ -58,7 +58,7 @@ async def test_warm_start_copies_base_with_decayed_visits(db_engine):
 
 @pytest.mark.asyncio
 async def test_cold_start_creates_warmstart_version(db_engine):
-    await _seed_base()
+    await _seed_base(A=11)                          # ktcommerce action_space=11 과 차원 일치해야 warm-start
     eng = await _reg().get_engine("ktcommerce")   # 활성 버전 없음 → cold-start
     policy, version_id, repo = await QTablePolicyStore.load(eng)
     err, ver = await repo.read(lambda s: repo.get_active_version(s))
@@ -67,6 +67,31 @@ async def test_cold_start_creates_warmstart_version(db_engine):
     assert policy.qtable.q[5, 2] == 0.9
 
 
+@pytest.mark.asyncio
+async def test_catalog_dim_change_migrates_preserving_learning(db_engine):
+    """카탈로그 카드 수 변경(9→11) 시 학습 보존 마이그레이션 — 겹치는 셀 복사 + 새 카드 fresh."""
+    import uuid as _uuid
+    cid = str(_uuid.uuid4())
+    # 이 회사 활성 버전을 A=9 로 시드 + 셀 (5,2)=0.9
+    repo = LearningRepository(cid)
+    vid = await repo.get_or_create_active_version(
+        state_space_size=162, action_space_size=9, learning_rate=0.1, discount_factor=0.95,
+        scope=2, version_name="old_v9")
+    await repo.upsert_cell(vid, state_index=5, action_id=2, q_value=0.9, count=7)
+
+    # 엔진(_base type:db → 카탈로그 11장) 로드 → 9≠11 감지 → 마이그레이션
+    eng = await _reg().get_engine(cid)
+    assert eng.action_space_size == 11
+    policy, new_vid, _ = await QTablePolicyStore.load(eng)
+    assert str(new_vid) != str(vid)            # 새 버전
+    assert policy.qtable.q[5, 2] == 0.9        # 기존 학습 보존
+    assert policy.qtable.q[5, 10] == 0.0       # 새 카드(action 10) fresh
+    assert policy.qtable.visits[5, 2] == 7     # 방문수도 보존
+    # 새 버전이 활성 · 차원 11
+    err, active = await repo.read(lambda s: repo.get_active_version(s))
+    assert str(active.version_id) == str(new_vid) and active.action_space_size == 11
+
+
 @pytest.mark.asyncio
 async def test_dimension_mismatch_falls_back_to_heuristic(db_engine):
     await _seed_base(S=162, A=9)

agent/tests/test_p7_apis.py

@@ -66,6 +66,26 @@ async def test_card_update_search(client, db_engine):
     assert any(m["card_id"] == "CUSTOM-X" for m in allm["mapping"])
 
 
+@pytest.mark.asyncio
+async def test_catalog_refresh(client, db_engine):
+    # 카탈로그 발행 후 엔진 재조립 트리거 — 성공 + action_space 반환. 헤더 없으면 400.
+    r = await client.post("/v1/catalog-refresh", headers=H)
+    assert r.status_code == 200
+    body = r.json()
+    assert body["success"] and body["action_space_size"] >= 1
+    r2 = await client.post("/v1/catalog-refresh")
+    assert r2.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_catalog_refresh_all_no_tenant_header(client, db_engine):
+    # 공용 카탈로그 전역 반영 — 테넌트 헤더 없이도 200(화이트리스트) + 캐시 클리어.
+    await client.post("/v1/catalog-refresh", headers=H)   # 엔진 하나 캐시
+    r = await client.post("/v1/catalog-refresh-all")       # 헤더 없음
+    assert r.status_code == 200
+    assert r.json()["success"] is True
+
+
 @pytest.mark.asyncio
 async def test_invalidate_and_reset_scoped(client, db_engine):
     # 가격협상 카드선택이 일어나는 긴 경로(850→900→990)로 양 테넌트 데이터 생성

agent/tests/test_p7_chat.py

@@ -125,24 +125,41 @@ def test_acceptance_ratio_dynamic_calc():
     assert calc({"input_price": 9800}) == 0.0                                # 기준가 전무(과거 세션 호환)
 
 
-def test_selected_card_mapping_overrides_tenant_default():
+def test_card_id_fixed_mapping_and_selection_mask():
-    """운영 세션은 견적 version_id 에 묶인 카드 번호를 action_id 순서로 사용한다."""
+    """카드 정리 후: action_id↔카드는 테넌트 매핑으로 고정, 견적 선택은 available_mask 로 걸러진다.
-    session = ChatSession(
+    (구 인덱스 방식 폐기 — selected[action_id] 인덱싱은 견적마다 action_id 의미가 달라져 Q-table 오염.)"""
-        session_id="00000000-0000-0000-0000-000000000001",
-        tenant_id="ktcommerce",
-        company_id="ktcommerce",
-        context={"selected_nego_card_numbers": ["2", "7"]},
-        action_space_size=2,
-    )
     class _Mapper:
-        def get_card_id(self, action_id):
+        _m = {i: f"NGC-A{i + 1:03d}" for i in range(11)}
-            return f"DEFAULT-{action_id}"
+        def get_card_id(self, a):
+            return self._m.get(a)
+        def get_action_id(self, num):
+            return next((a for a, c in self._m.items() if c == num), None)
     class _Engine:
         mapper = _Mapper()
+        action_space_size = 11
+    eng = _Engine()
+    session = ChatSession(
+        session_id="00000000-0000-0000-0000-000000000001", tenant_id="ktcommerce", company_id="ktcommerce",
+        context={"selected_nego_card_numbers": ["NGC-A003", "NGC-A008"]}, action_space_size=11,
+    )
 
-    assert ChatService._card_id_for_action(_Engine(), session, 0) == "2"
+    # ① card_id 는 고정 매핑 (선택 리스트 인덱싱 아님)
-    assert ChatService._card_id_for_action(_Engine(), session, 1) == "7"
+    assert ChatService._card_id_for_action(eng, session, 0) == "NGC-A001"
-    assert ChatService._card_id_for_action(_Engine(), session, 8) == "DEFAULT-8"
+    assert ChatService._card_id_for_action(eng, session, 2) == "NGC-A003"
+
+    # ② 선택은 mask 로 — NGC-A003(action 2), NGC-A008(action 7) 만 pickable
+    mask = ChatService._selection_mask(eng, session)
+    assert mask is not None and mask[2] and mask[7]
+    assert not mask[0] and not mask[5] and mask.sum() == 2
+
+    # ③ 사용한 action 은 mask 에서 제외
+    session.used_action_ids = {2}
+    mask2 = ChatService._selection_mask(eng, session)
+    assert not mask2[2] and mask2[7] and mask2.sum() == 1
+
+    # ④ 선택 없으면 None → 전체 허용(폴백)
+    session.context["selected_nego_card_numbers"] = []
+    assert ChatService._selection_mask(eng, session) is None
 
 
 def test_default_1pct_wildcard_still_runs_without_selected_wildcard():

negodata/backend/config/config_models.py

@@ -9,6 +9,7 @@ class WebServerConfig(ConfigModel):
     is_test: bool = False
     client_url: str = ""
     nego_chat_url: str = "http://localhost:3300"
+    agent_base_url: str = "http://localhost:9500"   # 협상 agent(9500). 공용 카탈로그 변경 알림용.
 
 
 class LogConfig(ConfigModel):

negodata/backend/services/agent_notify.py

@@ -0,0 +1,26 @@
+"""협상 agent 카탈로그 변경 알림 (best-effort).
+
+공용 협상카드(card.nego_cards, user_id NULL)는 agent 의 action space(Q-table action 축)를 정의한다.
+negodata 에서 공용 카드를 추가/삭제/번호변경하면 agent 의 캐시된 엔진을 무효화해 최신 카탈로그로
+재조립되게 알린다. 실패해도 카드 작업 자체는 성공으로 둔다 — agent 는 TTL/다음 재조립으로 결국 반영된다.
+
+개인(회사) 카드나 와일드카드는 action space 를 바꾸지 않으므로 알리지 않는다(호출부에서 판정).
+"""
+
+import httpx
+
+from common.logger import LOG
+from config.server_configs import web_server_config
+
+
+async def notify_catalog_changed() -> None:
+    if getattr(web_server_config, "is_test", False):
+        return  # 테스트 환경에선 외부 호출 안 함
+    base = (getattr(web_server_config, "agent_base_url", "") or "").rstrip("/")
+    if not base:
+        return
+    try:
+        async with httpx.AsyncClient(timeout=3.0) as cli:
+            await cli.post(f"{base}/v1/catalog-refresh-all")
+    except Exception as ex:  # 알림 실패는 카드 작업을 막지 않는다
+        LOG.w(f"[agent_notify] 공용 카탈로그 변경 알림 실패(무시): {ex}")

negodata/backend/services/card_service.py

@@ -7,6 +7,7 @@ from common.database.model.models import nego_cards, wild_cards
 from common.enums import CardStatus, DBWRType, ErrorType
 from common.models.gmodel import PageParams
 from crud.card_crud import ICardCRUD, CardCRUD
+from services.agent_notify import notify_catalog_changed
 from router.v1.card.protocol import CardData, Req_CreateCard, Req_UpdateCard, Res_Card, Res_CardList, Res_DeleteCard
 
 
@@ -196,6 +197,9 @@ class CardService:
         if err != ErrorType.SUCCESS:
             res.result.SetResult(err)
             return res
+        # 공용 일반카드는 agent action space 를 정의 → 변경 알림(전역 엔진 재조립). 개인/와일드는 미해당.
+        if req.is_shared and not is_wildcard:
+            await notify_catalog_changed()
         # 서버 기본값(created_at 등)은 insert 후 객체에 실리지 않으므로 재조회.
         return await self.get_card(user_id, str(getattr(card, pk_attr)))
 
@@ -210,6 +214,7 @@ class CardService:
         if err != ErrorType.SUCCESS:
             res.result.SetResult(err)
             return res
+        was_shared_nego = _row.user_id is None and not is_wild
 
         # 해당 테이블에 있는 컬럼만 추린다(없는 필드는 무시). status → available(와일드 전용).
         allowed = {"name", "number", "script", "edit_script", "usage_type"}
@@ -226,6 +231,9 @@ class CardService:
         if err != ErrorType.SUCCESS:
             res.result.SetResult(err)
             return res
+        # 공용 일반카드 변경(번호/개수 등)은 action space 에 영향 → 알림.
+        if was_shared_nego:
+            await notify_catalog_changed()
         return await self.get_card(user_id, card_id)
 
     # ---- 삭제(soft) ----------------------------------------------------------
@@ -238,6 +246,7 @@ class CardService:
         if err != ErrorType.SUCCESS:
             res.result.SetResult(err)
             return res
+        was_shared_nego = _row.user_id is None and not _is_wild
 
         err = await DB_SESSION_MNG.execute_lambda_run(
             [model.DBType()],
@@ -246,3 +255,7 @@ class CardService:
         if err != ErrorType.SUCCESS:
             res.result.SetResult(err)
             return res
+        # 공용 일반카드 삭제는 카탈로그 수 변경 → 알림.
+        if was_shared_nego:
+            await notify_catalog_changed()
+        return res

postgres-init/init-data/init-data.sql

@@ -54,7 +54,7 @@ INSERT INTO card.nego_cards
     (user_id, name, number, script, edit_script, usage_type, tone, strategy_type)
 SELECT * FROM (VALUES
     -- 1. 타사 검토 언급 — tone 1(강경) · strategy 1(경쟁)
-    (NULL::uuid, '타사 검토 언급', '1',
+    (NULL::uuid, '타사 검토 언급', 'NGC-001',
      '귀사의 품질과 협력 의지를 높이 평가하고 있어 우선적으로 협의를 진행하고 있습니다.
 
 다만 당사는 현재 본 건에 대해 복수의 공급 가능 업체와 병행하여 검토를 진행하고 있으며, 최종 선정을 위해서는 제시해 주신 조건이 타 대안 대비 분명한 경쟁력을 갖추어야 합니다.
@@ -64,7 +64,7 @@ SELECT * FROM (VALUES
      1, 1, 1),
 
     -- 2. 내부 승인 필요 — tone 2(정중) · strategy 1(경쟁)
-    (NULL::uuid, '내부 승인 필요', '2',
+    (NULL::uuid, '내부 승인 필요', 'NGC-002',
      '제안해 주신 조건은 잘 검토하였습니다.
 
 다만 해당 수준의 가격은 담당자 선에서 단독으로 확정하기 어려우며, 내부 상위 결재 절차를 거쳐야 하는 사안입니다. 솔직히 말씀드리면, 현재 조건으로는 내부 승인을 받기가 쉽지 않을 것으로 예상됩니다.
@@ -74,7 +74,7 @@ SELECT * FROM (VALUES
      1, 2, 1),
 
     -- 3. 장기 관계 강조 — tone 3(우호) · strategy 2(수용)
-    (NULL::uuid, '장기 관계 강조', '3',
+    (NULL::uuid, '장기 관계 강조', 'NGC-003',
      '귀사와의 협력을 진심으로 소중하게 생각하고 있습니다.
 
 이번 거래가 단발성으로 끝나지 않고, 앞으로 오랜 기간 함께할 관계의 시작이 되기를 바랍니다. 그런 의미에서 양사가 모두 만족할 수 있는 합리적인 지점을 함께 찾고자 합니다.
@@ -84,7 +84,7 @@ SELECT * FROM (VALUES
      1, 3, 2),
 
     -- 4. 명분 제공 — tone 2(정중) · strategy 2(수용)
-    (NULL::uuid, '명분 제공', '4',
+    (NULL::uuid, '명분 제공', 'NGC-004',
      '그동안 귀사께서 보여주신 협상 태도와 전문성을 높이 평가하고 있습니다.
 
 이번 조정은 귀사가 일방적으로 양보하는 것이 아니라, 양사가 장기적 협력을 위해 함께 내린 전략적 결정으로 이해해 주시면 감사하겠습니다. 당사 역시 이번 합의를 귀사와의 신뢰를 보여주는 기회로 삼겠습니다.
@@ -94,7 +94,7 @@ SELECT * FROM (VALUES
      1, 2, 2),
 
     -- 5. 공정 합의 제안 — tone 4(중립) · strategy 2(수용)
-    (NULL::uuid, '공정 합의 제안', '5',
+    (NULL::uuid, '공정 합의 제안', 'NGC-005',
      '당사가 바라는 것은 어느 한쪽에 치우친 거래가 아니라, 양사 모두가 공정하다고 느낄 수 있는 합의입니다.
 
 당사는 합리적인 근거 위에서 성실하게 조건을 제시해 왔으며, 귀사 역시 같은 자세로 임해 주시리라 믿습니다. 서로가 공정함을 기준으로 한 걸음씩 다가선다면, 양사 모두 납득할 수 있는 결론에 이를 수 있습니다.
@@ -104,7 +104,7 @@ SELECT * FROM (VALUES
      1, 4, 2),
 
     -- 6. 총비용 가치 설명 — tone 4(중립) · strategy 3(고수)
-    (NULL::uuid, '총비용 가치 설명', '6',
+    (NULL::uuid, '총비용 가치 설명', 'NGC-006',
      '제안 드린 가격에 대해 부담을 느끼시는 점 충분히 이해합니다.
 
 다만 당사의 제안 가격은 단순한 단가가 아니라 안정적인 품질과 납기 준수, 사후 지원까지 포함한 총비용 관점에서 산정된 것입니다. 초기 단가만을 기준으로 비교할 경우, 운영 과정에서 발생할 수 있는 추가 비용이나 리스크가 충분히 반영되지 않을 수 있습니다.
@@ -114,7 +114,7 @@ SELECT * FROM (VALUES
      1, 4, 3),
 
     -- 7. 예산 상한 안내 — tone 2(정중) · strategy 3(고수)
-    (NULL::uuid, '예산 상한 안내', '7',
+    (NULL::uuid, '예산 상한 안내', 'NGC-007',
      '귀사와의 합의를 진심으로 바라고 있습니다.
 
 다만 당사 내부 예산 정책상 본 건에 책정 가능한 금액은 {anchoring_price}원(VAT별도)이 한계입니다. 이는 개인의 재량을 넘어선 내부 기준에 해당하여 조정이 어려운 부분이니 너른 양해를 부탁드립니다.
@@ -124,7 +124,7 @@ SELECT * FROM (VALUES
      1, 2, 3),
 
     -- 8. 시장가 근거 제시 — tone 4(중립) · strategy 3(고수)
-    (NULL::uuid, '시장가 근거 제시', '8',
+    (NULL::uuid, '시장가 근거 제시', 'NGC-008',
      '당사의 제안 가격이 시장에서 동떨어진 수준이 아님을 함께 확인해 보고자 합니다.
 
 인터넷 시장 조사 결과, 동종 업계의 유사 거래에서도 본 건과 비슷한 사양은 {internet_lowest_price}원(VAT별도) 안팎에서 합의되고 있습니다. 당사의 제안은 이러한 시장의 일반적인 수준을 충실히 반영한 것입니다.
@@ -134,7 +134,7 @@ SELECT * FROM (VALUES
      1, 4, 3),
 
     -- 9. 조건부 가격 조정 — tone 4(중립) · strategy 4(협력)
-    (NULL::uuid, '조건부 가격 조정', '9',
+    (NULL::uuid, '조건부 가격 조정', 'NGC-009',
      '귀사의 입장을 고려하여 당사가 한 걸음 더 나아가고자 합니다.
 
 귀사께서 {customer_condition}에 동의해 주신다면, 당사는 가격을 {target_price}원(VAT별도)까지 조정하겠습니다.
@@ -144,7 +144,7 @@ SELECT * FROM (VALUES
      1, 4, 4),
 
     -- 10. 향후 거래 연계 — tone 4(중립) · strategy 4(협력)
-    (NULL::uuid, '향후 거래 연계', '10',
+    (NULL::uuid, '향후 거래 연계', 'NGC-010',
      '이번 거래의 가격을 {target_price}원(VAT별도)으로 조정하는 대신, {customer_condition}을 함께 검토해 주실 것을 제안 드립니다.
 
 당장의 단가 한 건만 보기보다 향후 이어질 거래까지 함께 고려한다면, 양사 모두에게 더 큰 가치를 만들 수 있습니다. 이번 합의를 장기적 관계의 출발점으로 삼아, 서로에게 이익이 되는 구조를 함께 설계하기를 바랍니다.',
@@ -152,7 +152,7 @@ SELECT * FROM (VALUES
      1, 4, 4),
 
     -- 11. 양보 가치 강조 — tone 4(중립) · strategy 4(협력)
-    (NULL::uuid, '양보 가치 강조', '11',
+    (NULL::uuid, '양보 가치 강조', 'NGC-011',
      '이번 조정은 당사에 결코 작은 일이 아님을 먼저 말씀드리고 싶습니다.
 
 {target_price}원(VAT별도)으로의 조정은 당사 수익 구조상 상당한 부담을 감수한 결정이며, 내부적으로도 쉽지 않은 승인 과정을 거쳤습니다. 이는 곧 귀사께서 그만큼 실질적인 혜택을 얻으시게 된다는 의미이기도 합니다.
@@ -168,7 +168,7 @@ INSERT INTO card.wild_cards
     (user_id, name, number, script, edit_script, usage_type, condition, available, memo, tone, strategy_type)
 SELECT * FROM (VALUES
     -- 1. 목표가 선제안 — tone 5(단호) · strategy 5(선점)
-    (NULL::uuid, '목표가 선제안', '1',
+    (NULL::uuid, '목표가 선제안', 'WC-01',
      '안녕하십니까. 금번 협상에 참여해 주셔서 감사합니다.
 
 당사는 {customer_reference}을(를) 종합적으로 검토하여 합리적인 목표 가격을 산정하였으며, 이에 {target_price}원(VAT별도)을 제안 드립니다.
@@ -178,7 +178,7 @@ SELECT * FROM (VALUES
      1, NULL::varchar, TRUE, NULL::varchar, 5, 5),
 
     -- 2. 역제안가 제시 — tone 5(단호) · strategy 5(선점)
-    (NULL::uuid, '역제안가 제시', '2',
+    (NULL::uuid, '역제안가 제시', 'WC-02',
      '제안해 주신 {prev_partner_price}원은 당사가 검토한 기준 대비 다소 높은 수준으로 판단됩니다.
 
 이에 당사는 {target_mid_price}원(VAT별도)을 역으로 제안 드립니다. 양측 제안 사이에서 합리적인 접점을 찾되, 그 기준은 명확한 산정 근거에 두는 것이 바람직하다고 봅니다.
@@ -188,7 +188,7 @@ SELECT * FROM (VALUES
      1, NULL::varchar, TRUE, NULL::varchar, 5, 5),
 
     -- 3. 최종 통보 — tone 1(강경) · strategy 1(경쟁)
-    (NULL::uuid, '최종 통보', '3',
+    (NULL::uuid, '최종 통보', 'WC-03',
      '합리적인 기준에 근거하여 목표 가격을 제안 드렸으나, 귀사의 기존 제안 가격으로는 긍정적인 합의가 어려울 것으로 예상됩니다.
 
 이번 협상이 결렬되는 경우 우선 협상권을 보장하기 어려우며, 다른 공급 업체를 선정하기 위한 검토가 진행될 수 있습니다.
@@ -198,7 +198,7 @@ SELECT * FROM (VALUES
      1, NULL::varchar, TRUE, NULL::varchar, 1, 1),
 
     -- 4. 단계적 인하 제안 — tone 1(강경) · strategy 1(경쟁)
-    (NULL::uuid, '단계적 인하 제안', '4',
+    (NULL::uuid, '단계적 인하 제안', 'WC-04',
      '당초 당사가 검토한 적정가는 {anchoring_price}원(VAT별도) 수준이었습니다. 다만 귀사의 입장과 시장 상황을 함께 고려하여, 당사가 한발 물러서고자 합니다.
 
 이에 {target_price}원(VAT별도)으로 조정하여 제안 드립니다. 이는 당초 기준 대비 당사가 상당 부분 양보한 금액이니, 귀사께서도 이 점을 감안하여 긍정적으로 검토해 주시기를 부탁드립니다.',
@@ -206,7 +206,7 @@ SELECT * FROM (VALUES
      1, NULL::varchar, TRUE, NULL::varchar, 1, 1),
 
     -- 5. 중간값 절충 — tone 4(중립) · strategy 6(종결)
-    (NULL::uuid, '중간값 절충', '5',
+    (NULL::uuid, '중간값 절충', 'WC-05',
      '긴 협의 끝에 양측의 입장이 상당히 가까워졌습니다.
 
 현재 당사 제안 {prev_customer_price}원과 귀사 제안 {prev_partner_price}원 사이의 차이를 양사가 절반씩 나누어, {middle_price}원(VAT별도)으로 마무리할 것을 제안 드립니다.