feat(backend): 협상 세션 목록 + 참여 기능

- GET /v1/negotiation/sessions: 로그인 공급사의 세션 목록(필터/정렬/페이지네이션)
  · qt_end_time 은 견적(quotation.end_time) 기준, sessions⨝items⨝quotations 조인
  · status/qt_type 은 정수 코드로 응답(라벨 매핑은 프론트)
- POST /v1/negotiation/sessions/{session_id}/participate: 협상 참여
  · 검증: 소유(공급사 대조)→세션상태→견적마감→마감시간, 에러코드 1300~1304
  · 협상생성→협상중, 견적→견적진행중 (협상중/완료는 무변경 진입)
  · 마감초과 시 협상생성 세션만 미참여로 정리
- DBType.NEGOTIATION/QUOTATION, items/sessions/quotations 모델
- QtType/SessionStatus/QuotationStatus enum, AuthService.authenticate 공통화
- 협상 e2e 테스트(test_negotiation.py)

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

backend/common/database/db_session_manager.py

@@ -34,21 +34,27 @@ class DBSessionManager(Singleton):
         # 종료 시 dispose 하기 위해 생성한 엔진을 모아둔다.
         self.__engines = []
         # 논리 DB -> config. DB 가 늘어나면 여기에 추가만 하면 된다.
-        # USER/PARTNER 는 물리적으로 같은 negosium_db 라 main_db_config 를 재사용한다(도메인별 논리 구분용).
+        # USER/PARTNER/NEGOTIATION/QUOTATION 은 물리적으로 같은 negosium_db 라 main_db_config 를 재사용한다(도메인별 논리 구분용).
         self.__db_type_map = {
             DBType.USER.value: main_db_config,
             DBType.PARTNER.value: main_db_config,
+            DBType.NEGOTIATION.value: main_db_config,
+            DBType.QUOTATION.value: main_db_config,
         }
 
         # Write 엔진 맵
         self.__write_session = {
             DBType.USER.value: self.create_engine(DBType.USER.value, DBWRType.DB_WRITE.value),
             DBType.PARTNER.value: self.create_engine(DBType.PARTNER.value, DBWRType.DB_WRITE.value),
+            DBType.NEGOTIATION.value: self.create_engine(DBType.NEGOTIATION.value, DBWRType.DB_WRITE.value),
+            DBType.QUOTATION.value: self.create_engine(DBType.QUOTATION.value, DBWRType.DB_WRITE.value),
         }
         # Read 엔진 맵
         self.__read_session = {
             DBType.USER.value: self.create_engine(DBType.USER.value, DBWRType.DB_READ.value),
             DBType.PARTNER.value: self.create_engine(DBType.PARTNER.value, DBWRType.DB_READ.value),
+            DBType.NEGOTIATION.value: self.create_engine(DBType.NEGOTIATION.value, DBWRType.DB_READ.value),
+            DBType.QUOTATION.value: self.create_engine(DBType.QUOTATION.value, DBWRType.DB_READ.value),
         }
 
     def create_engine(self, db_type: int, db_wr_type: int):

backend/common/database/model/models.py

@@ -1,5 +1,5 @@
 from sqlalchemy.orm import declarative_base
-from sqlalchemy import Column, Integer, String, Boolean, DateTime, SmallInteger
+from sqlalchemy import Column, Integer, String, Boolean, DateTime, SmallInteger, BigInteger
 from sqlalchemy.dialects.postgresql import UUID, JSONB
 from sqlalchemy.sql import text
 
@@ -57,6 +57,104 @@ class suppliers(MAIN_BASE):
     deleted = Column(Boolean, nullable=False, server_default=text("false"))                                          # 소프트 삭제 여부
 
 
+class items(MAIN_BASE):
+    # partner.items (상품).
+    @staticmethod
+    def DBType():
+        return DBType.PARTNER.value
+
+    __tablename__ = "items"
+    __table_args__ = {"schema": "partner"}
+
+    item_id = Column(UUID(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()"))                 # 상품 식별자(PK)
+    company_id = Column(UUID(as_uuid=True), nullable=False)                                                          # 소속 회사(company.companies.company_id)
+    user_id = Column(UUID(as_uuid=True), nullable=False)                                                             # 등록 유저(company.users.user_id)
+    name = Column(String(100), nullable=False)                                                                       # 상품명
+    code = Column(String(30), nullable=True)                                                                         # 상품 코드
+    price = Column(BigInteger, nullable=True)                                                                        # 가격(원)
+    category = Column(String(255), nullable=True)                                                                    # 카테고리
+    image_url = Column(String(255), nullable=True)                                                                   # 이미지 URL
+    model_name = Column(String(100), nullable=True)                                                                  # 모델명
+    spec = Column(String(255), nullable=True)                                                                        # 규격
+    moq = Column(String(50), nullable=True)                                                                          # 최소 주문 수량
+    lead_time = Column(SmallInteger, nullable=True)                                                                  # 배송 리드타임
+    manufacturer = Column(String(50), nullable=True)                                                                 # 제조사
+    made_in = Column(String(100), nullable=True)                                                                     # 원산지
+    quantity_unit = Column(SmallInteger, nullable=True)                                                              # 취급 단위 (코드, 앱 enum 매핑)
+    delivery_type = Column(SmallInteger, nullable=True)                                                              # 배송 유형 (코드, 앱 enum 매핑)
+    vat_yn = Column(Boolean, nullable=True)                                                                          # 부가세 포함 여부
+    delivery_fee_yn = Column(Boolean, nullable=True)                                                                 # 배송비 포함 여부
+    internet_lowest_price_yn = Column(Boolean, nullable=False, server_default=text("false"))                         # 최저가 솔루션 보조 컬럼
+    category_type = Column(Integer, nullable=False, server_default=text("1"))                                        # 카테고리 조회용 자동 증가 숫자
+    created_at = Column(DateTime(timezone=True), nullable=False, server_default=text("(now() AT TIME ZONE 'utc')"))  # 생성 시각(UTC)
+    updated_at = Column(DateTime(timezone=True), nullable=False, server_default=text("(now() AT TIME ZONE 'utc')"))  # 수정 시각(UTC, 앱에서 갱신)
+    deleted = Column(Boolean, nullable=False, server_default=text("false"))                                          # 소프트 삭제 여부
+
+
+class sessions(MAIN_BASE):
+    # negotiation.sessions (협상 세션).
+    @staticmethod
+    def DBType():
+        return DBType.NEGOTIATION.value
+
+    __tablename__ = "sessions"
+    __table_args__ = {"schema": "negotiation"}
+
+    session_id = Column(UUID(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()"))              # 협상 세션 식별자(PK)
+    quotation_id = Column(UUID(as_uuid=True), nullable=False)                                                        # 소속 견적(quotation.quotations.qt_id)
+    item_id = Column(UUID(as_uuid=True), nullable=False)                                                             # 대상 상품(partner.items.item_id)
+    supplier_id = Column(UUID(as_uuid=True), nullable=False)                                                         # 대상 공급사(partner.suppliers.supplier_id)
+    qt_number = Column(String(30), nullable=False)                                                                   # 견적번호(스냅샷)
+    qt_round = Column(Integer, nullable=False)                                                                       # 견적 라운드(스냅샷)
+    qt_type = Column(SmallInteger, nullable=False)                                                                   # 견적 유형: 1=재협상, 2=재견적 (QtType)
+    target_price = Column(BigInteger, nullable=False)                                                                # 목표가(원)
+    status = Column(SmallInteger, nullable=False)                                                                    # 진행 상태 (SessionStatus 코드)
+    bid_price = Column(BigInteger, nullable=True)                                                                    # 입찰가(원)
+    bid_at = Column(DateTime(timezone=True), nullable=True)                                                          # 입찰 시각
+    end_time = Column(DateTime(timezone=True), nullable=False)                                                       # 세션 종료(마감) 시각
+    reject_reason = Column(String(255), nullable=True)                                                               # 거절 사유
+    reject_price = Column(BigInteger, nullable=True)                                                                 # 거절 시 제시가(원)
+    reject_delivery_type = Column(SmallInteger, nullable=True)                                                       # 거절 시 배송 유형 (코드)
+    created_at = Column(DateTime(timezone=True), nullable=False, server_default=text("(now() AT TIME ZONE 'utc')"))  # 생성 시각(UTC)
+    updated_at = Column(DateTime(timezone=True), nullable=False, server_default=text("(now() AT TIME ZONE 'utc')"))  # 수정 시각(UTC, 앱에서 갱신)
+    deleted = Column(Boolean, nullable=False, server_default=text("false"))                                          # 소프트 삭제 여부
+
+
+class quotations(MAIN_BASE):
+    # quotation.quotations (견적).
+    @staticmethod
+    def DBType():
+        return DBType.QUOTATION.value
+
+    __tablename__ = "quotations"
+    __table_args__ = {"schema": "quotation"}
+
+    qt_id = Column(UUID(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()"))                   # 견적 식별자(PK)
+    user_id = Column(UUID(as_uuid=True), nullable=False)                                                             # 생성 유저(company.users.user_id)
+    qt_setting_id = Column(UUID(as_uuid=True), nullable=False)                                                       # 견적 설정(quotation.quotation_settings.qt_setting_id)
+    version_id = Column(UUID(as_uuid=True), nullable=False)                                                          # 버전(card.versions.version_id)
+    name = Column(String(50), nullable=False)                                                                        # 견적명
+    number = Column(String(30), nullable=False)                                                                      # 견적번호
+    type = Column(SmallInteger, nullable=False)                                                                      # 견적 유형: 1=재협상, 2=재견적 (QtType)
+    round = Column(Integer, nullable=False, server_default=text("1"))                                                # 재견적 회차
+    status = Column(SmallInteger, nullable=False)                                                                    # 진행 상태 (QuotationStatus 코드)
+    start_time = Column(DateTime(timezone=True), nullable=False)                                                     # 견적 시작 시각
+    end_time = Column(DateTime(timezone=True), nullable=False)                                                       # 견적 종료(마감) 시각
+    manager_name = Column(String(50), nullable=True)                                                                 # 담당자명
+    manager_email = Column(String(255), nullable=True)                                                               # 담당자 이메일
+    manager_contact_number = Column(String(20), nullable=True)                                                       # 담당자 연락처
+    memo = Column(String(100), nullable=True)                                                                        # 메모
+    iteration = Column(Integer, nullable=False, server_default=text("0"))                                            # 반복 횟수
+    preferred_sp_yn = Column(Boolean, nullable=True)                                                                 # 선호 공급사 지정 여부
+    preferred_sp_id = Column(UUID(as_uuid=True), nullable=True)                                                      # 선호 공급사(partner.suppliers.supplier_id)
+    preferred_sp_name = Column(String(20), nullable=True)                                                            # 선호 공급사명(스냅샷)
+    equal_bid_yn = Column(Boolean, nullable=True)                                                                    # 동일가 입찰 발생 여부
+    equal_bid_data = Column(JSONB, nullable=True)                                                                    # 동일가 입찰 상세(JSON)
+    created_at = Column(DateTime(timezone=True), nullable=False, server_default=text("(now() AT TIME ZONE 'utc')"))  # 생성 시각(UTC)
+    updated_at = Column(DateTime(timezone=True), nullable=False, server_default=text("(now() AT TIME ZONE 'utc')"))  # 수정 시각(UTC, 앱에서 갱신)
+    deleted = Column(Boolean, nullable=False, server_default=text("false"))                                          # 소프트 삭제 여부
+
+
 class supplier_user_tokens(MAIN_BASE):
     # 유저 인증 토큰. supplier_users 1 : N tokens.
     @staticmethod

backend/common/enums.py

@@ -37,6 +37,13 @@ class ErrorType(Enum):
     ACCOUNT_BLOCKED_USER = auto()
     TOKEN_REVOKED = auto()  # 제시된 토큰이 저장된 토큰과 불일치(로그아웃/타기기 로그인으로 교체됨)
 
+    # 협상(negotiation) 관련 에러 — 프론트 toast 용 코드
+    NEGO_FORBIDDEN = 1300         # 공급사 불일치(권한 없음)
+    NEGO_NOT_PARTICIPABLE = auto()  # 1301 세션 상태가 미참여/협상거부라 참여 불가
+    NEGO_QUOTATION_CLOSED = auto()  # 1302 견적 마감 상태
+    NEGO_DEADLINE_PASSED = auto()   # 1303 견적 마감 시간 초과
+    NEGO_NOT_FOUND = auto()         # 1304 세션/견적 없음
+
 
 # ErrorType 의 HTTP_* 값과 status_code 를 맞춰 router 단에서 raise 한다.
 EXCEPTION_INVALID_CLIENT_REQUEST = HTTPException(status_code=ErrorType.HTTP_INVALID_CLIENT_REQUEST.value, detail=ErrorType.HTTP_INVALID_CLIENT_REQUEST.name)
@@ -53,8 +60,10 @@ class DBType(Enum):
     물리적으로 같은 negosium_db 라도 도메인별 논리 구분으로 나눠 둘 수 있다(커넥션 config 는 재사용).
     """
 
-    USER = 1     # 기본 유저 (supplier_users 테이블)
+    USER = 1         # 기본 유저 (supplier_users 테이블)
-    PARTNER = 2  # partner 도메인 (partner.suppliers 등)
+    PARTNER = 2      # partner 도메인 (partner.suppliers, partner.items 등)
+    NEGOTIATION = 3  # negotiation 도메인 (negotiation.sessions 등)
+    QUOTATION = 4    # quotation 도메인 (quotation.quotations 등)
 
 
 class DBWRType(Enum):
@@ -87,3 +96,32 @@ class TokenType(Enum):
 
     ACCESS = 1
     REFRESH = 2
+
+
+class QtType(Enum):
+    """견적/세션 유형 코드. quotation.quotations.type / negotiation.sessions.qt_type."""
+
+    RENEGO = 1    # 재협상(1:1)
+    REQUOTE = 2   # 재견적(1:N)
+
+
+class SessionStatus(Enum):
+    """협상 세션 진행 상태 코드. negotiation.sessions.status.
+    ⚠️ 세션을 생성/갱신하는 쪽(바이어/agent)과 코드값이 일치해야 한다.
+    """
+
+    CREATED = 1           # 협상생성
+    IN_PROGRESS = 2       # 협상중
+    DONE = 3              # 협상완료
+    NOT_PARTICIPATED = 4  # 미참여
+    REJECTED = 5          # 협상거부
+
+
+class QuotationStatus(Enum):
+    """견적 진행 상태 코드. quotation.quotations.status.
+    ⚠️ 견적을 생성/갱신하는 쪽(바이어/agent)과 코드값이 일치해야 한다.
+    """
+
+    CREATED = 1      # 견적생성
+    IN_PROGRESS = 2  # 견적진행중
+    CLOSED = 3       # 견적마감

backend/crud/session_crud.py

@@ -0,0 +1,140 @@
+from abc import ABC, abstractmethod
+from typing import Tuple
+
+from sqlalchemy import asc, desc, func, select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.database.model.models import items, quotations, sessions
+from common.enums import ErrorType
+from common.logger import LOG
+
+
+# 협상 세션 CRUD. 목록은 세션(negotiation) ⨝ 상품(partner) ⨝ 견적(quotation) 조인으로 만든다.
+# 마감일(qt_end_time)은 견적(quotation.end_time)이 진실값이다(session.end_time 은 협상 종료 시점 기록용).
+class ISessionCRUD(ABC):
+    @abstractmethod
+    async def list_by_supplier(self, cdb: AsyncSession, supplier_id, status, qt_type, order, offset, limit) -> Tuple[ErrorType, list]:
+        pass
+
+    @abstractmethod
+    async def count_by_supplier(self, cdb: AsyncSession, supplier_id, status, qt_type) -> Tuple[ErrorType, int]:
+        pass
+
+    @abstractmethod
+    async def get_session_by_id(self, cdb: AsyncSession, session_id) -> Tuple[ErrorType, sessions]:
+        pass
+
+    @abstractmethod
+    async def get_quotation_by_id(self, cdb: AsyncSession, quotation_id) -> Tuple[ErrorType, quotations]:
+        pass
+
+    @abstractmethod
+    async def update_session_status(self, cdb: AsyncSession, session_id, status: int) -> ErrorType:
+        pass
+
+    @abstractmethod
+    async def update_quotation_status(self, cdb: AsyncSession, quotation_id, status: int) -> ErrorType:
+        pass
+
+
+class SessionCRUD(ISessionCRUD):
+    @staticmethod
+    def __filters(supplier_id, status, qt_type):
+        conds = [sessions.supplier_id == supplier_id, sessions.deleted == False]  # noqa: E712
+        if status is not None:
+            conds.append(sessions.status == status)
+        if qt_type is not None:
+            conds.append(sessions.qt_type == qt_type)
+        return conds
+
+    async def list_by_supplier(self, cdb: AsyncSession, supplier_id, status, qt_type, order, offset, limit) -> Tuple[ErrorType, list]:
+        try:
+            conds = self.__filters(supplier_id, status, qt_type)
+            order_col = desc(quotations.end_time) if order == "desc" else asc(quotations.end_time)
+            query = (
+                select(
+                    sessions.session_id,
+                    sessions.status,
+                    sessions.qt_type,
+                    sessions.qt_number,
+                    quotations.end_time,  # qt_end_time = 견적 마감 시각
+                    items.code,
+                    items.name,
+                    items.model_name,
+                    items.manufacturer,
+                )
+                .join(items, items.item_id == sessions.item_id)
+                .join(quotations, quotations.qt_id == sessions.quotation_id)
+                .where(*conds, items.deleted == False, quotations.deleted == False)  # noqa: E712
+                .order_by(order_col)
+                .offset(offset)
+                .limit(limit)
+            )
+            err_type, rows = await DB_SESSION_MNG.execute(cdb, query, "list_by_supplier failed.")
+            if err_type != ErrorType.SUCCESS:
+                return err_type, []
+            return ErrorType.SUCCESS, rows
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED, []
+
+    async def count_by_supplier(self, cdb: AsyncSession, supplier_id, status, qt_type) -> Tuple[ErrorType, int]:
+        try:
+            conds = self.__filters(supplier_id, status, qt_type)
+            query = (
+                select(func.count())
+                .select_from(sessions)
+                .join(items, items.item_id == sessions.item_id)
+                .join(quotations, quotations.qt_id == sessions.quotation_id)
+                .where(*conds, items.deleted == False, quotations.deleted == False)  # noqa: E712
+            )
+            err_type, rows = await DB_SESSION_MNG.execute(cdb, query, "count_by_supplier failed.")
+            if err_type != ErrorType.SUCCESS:
+                return err_type, 0
+            return ErrorType.SUCCESS, (rows[0] if rows else 0)
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED, 0
+
+    async def get_session_by_id(self, cdb: AsyncSession, session_id) -> Tuple[ErrorType, sessions]:
+        try:
+            query = select(sessions).where(sessions.session_id == session_id, sessions.deleted == False).limit(1)  # noqa: E712
+            err_type, row_list = await DB_SESSION_MNG.execute(cdb, query, f"get_session_by_id({session_id}) failed.")
+            if err_type != ErrorType.SUCCESS:
+                return err_type, None
+            if len(row_list) != 1:
+                return ErrorType.DB_INVALID_KEY, None
+            return ErrorType.SUCCESS, row_list[0]
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED, None
+
+    async def get_quotation_by_id(self, cdb: AsyncSession, quotation_id) -> Tuple[ErrorType, quotations]:
+        try:
+            query = select(quotations).where(quotations.qt_id == quotation_id, quotations.deleted == False).limit(1)  # noqa: E712
+            err_type, row_list = await DB_SESSION_MNG.execute(cdb, query, f"get_quotation_by_id({quotation_id}) failed.")
+            if err_type != ErrorType.SUCCESS:
+                return err_type, None
+            if len(row_list) != 1:
+                return ErrorType.DB_INVALID_KEY, None
+            return ErrorType.SUCCESS, row_list[0]
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED, None
+
+    async def update_session_status(self, cdb: AsyncSession, session_id, status: int) -> ErrorType:
+        try:
+            query = update(sessions).where(sessions.session_id == session_id).values(status=status)
+            return await DB_SESSION_MNG.add(cdb, query)
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED
+
+    async def update_quotation_status(self, cdb: AsyncSession, quotation_id, status: int) -> ErrorType:
+        try:
+            query = update(quotations).where(quotations.qt_id == quotation_id).values(status=status)
+            return await DB_SESSION_MNG.add(cdb, query)
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED

backend/router/router.py

@@ -10,6 +10,7 @@ from common.logger import LOG
 from common.utils.gtime import GTime
 from config.server_configs import web_server_config
 import router.v1.auth.account
+import router.v1.negotiation.session
 
 API_SERVER_START_TIME = GTime.UTCStr()
 
@@ -55,3 +56,4 @@ async def healthz():
 
 # 각 도메인 라우터를 등록한다. 새 기능 추가 시 router.v1.<domain>.<file> 를 import 후 include.
 app.include_router(router.v1.auth.account.router)
+app.include_router(router.v1.negotiation.session.router)

backend/router/v1/negotiation/protocol.py

@@ -0,0 +1,25 @@
+from common.models.gmodel import Res_WebPacketProtocol, WebPacketProtocol
+
+
+# 협상 세션 목록 행. status/qt_type 은 정수 코드로 내려가고 라벨 매핑은 프론트가 한다.
+class ListItem(WebPacketProtocol):
+    session_id: str = ""
+    session_status: int = 0   # SessionStatus 코드
+    qt_type: int = 0          # QtType 코드 (1=재협상, 2=재견적)
+    qt_number: str = ""
+    qt_end_time: str = ""     # ISO 8601 (마감 시각)
+    item_code: str = ""
+    item_name: str = ""
+    model_name: str = ""
+    maker_name: str = ""
+
+
+class Res_SessionList(Res_WebPacketProtocol):
+    items: list[ListItem] = []
+    total: int = 0
+    page: int = 0
+    page_size: int = 0
+
+
+class Res_Participate(Res_WebPacketProtocol):
+    session_id: str = ""  # 참여 성공한 세션 (채팅 진입용)

backend/router/v1/negotiation/session.py

@@ -0,0 +1,47 @@
+from typing import Optional
+
+from fastapi import APIRouter, Depends, Query
+from fastapi.security import HTTPAuthorizationCredentials
+
+from common.models.gmodel import UserInfo
+from router.v1.validator.dependencies import IsValidAccessToken, RemoveNoneResponse, security
+from services.negotiation_service import NegotiationService
+from .protocol import Res_Participate, Res_SessionList
+
+router = APIRouter(prefix="/v1/negotiation", tags=["Negotiation"], responses={404: {"description": "Not found"}})
+
+
+@router.get(
+    path="/sessions",
+    response_model=Res_SessionList,
+    summary="협상 세션 목록",
+    description="로그인한 공급사의 협상 세션 목록. 필터(status/qt_type, 정수 코드)·마감일 정렬·페이지네이션 지원.",
+)
+async def list_sessions(
+    user_info: UserInfo = Depends(IsValidAccessToken),
+    credentials: HTTPAuthorizationCredentials = Depends(security),
+    service: NegotiationService = Depends(),
+    status: Optional[int] = Query(None, description="세션 상태 코드 (SessionStatus)"),
+    qt_type: Optional[int] = Query(None, description="견적 유형 코드 (QtType: 1=재협상, 2=재견적)"),
+    order: str = Query("asc", description="마감일 정렬: asc(임박순)/desc"),
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+):
+    return RemoveNoneResponse(
+        await service.list_sessions(user_info, credentials.credentials, status, qt_type, order, page, page_size)
+    )
+
+
+@router.post(
+    path="/sessions/{session_id}/participate",
+    response_model=Res_Participate,
+    summary="협상 참여",
+    description="세션에 참여한다. 소유(공급사)·세션상태·견적마감·마감시간 검증 후 협상생성→협상중, 견적→견적진행중으로 전이.",
+)
+async def participate(
+    session_id: str,
+    user_info: UserInfo = Depends(IsValidAccessToken),
+    credentials: HTTPAuthorizationCredentials = Depends(security),
+    service: NegotiationService = Depends(),
+):
+    return RemoveNoneResponse(await service.participate(user_info, credentials.credentials, session_id))

backend/services/auth_service.py

@@ -218,15 +218,23 @@ class AuthService:
         )
         return err_type == ErrorType.SUCCESS and stored == presented
 
-    async def get_me(self, user_info: UserInfo, access_token: str) -> Res_Me:
+    async def authenticate(self, user_info: UserInfo, access_token: str) -> tuple[ErrorType, UserInfo]:
-        # 토큰 디코드는 라우터 Depends(IsValidAccessToken) 에서 수행됨. 여기선 su_id DB 검증 + 저장 토큰 대조.
+        """access 토큰 보호 요청 공통 인증: 계정 활성 확인 + 저장된 access 토큰 대조.
-        res = Res_Me()
+        성공 시 (SUCCESS, DB 최신 UserInfo), 실패 시 (에러코드, None). 다른 도메인 service 에서도 재사용한다.
+        """
         err_type, info = await self.__load_active_account(user_info.su_id)
         if err_type != ErrorType.SUCCESS:
-            res.result.SetResult(err_type)
+            return err_type, None
-            return res
         if not await self.__verify_stored_token(info.su_id, TokenType.ACCESS.value, access_token):
-            res.result.SetResult(ErrorType.TOKEN_REVOKED)  # 로그아웃/타기기 로그인으로 무효화됨
+            return ErrorType.TOKEN_REVOKED, None  # 로그아웃/타기기 로그인으로 무효화됨
+        return ErrorType.SUCCESS, info
+
+    async def get_me(self, user_info: UserInfo, access_token: str) -> Res_Me:
+        # 토큰 디코드는 라우터 Depends(IsValidAccessToken) 에서 수행됨. 여기선 공통 인증으로 검증.
+        res = Res_Me()
+        err_type, info = await self.authenticate(user_info, access_token)
+        if err_type != ErrorType.SUCCESS:
+            res.result.SetResult(err_type)
             return res
         res.su_id = info.su_id
         res.id = info.id

backend/services/negotiation_service.py

@@ -0,0 +1,150 @@
+import uuid
+from datetime import datetime, timezone
+
+from fastapi import Depends
+
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.database.model.models import sessions
+from common.enums import DBWRType, ErrorType, QuotationStatus, SessionStatus
+from common.models.gmodel import UserInfo
+from crud.session_crud import ISessionCRUD, SessionCRUD
+from router.v1.negotiation.protocol import ListItem, Res_Participate, Res_SessionList
+from services.auth_service import AuthService
+
+
+class NegotiationService:
+    """협상 도메인 비즈니스 로직.
+    - 인증(계정 활성 + 저장 토큰 대조)은 AuthService.authenticate 로 위임(재사용).
+    - 목록은 로그인 유저의 supplier_id 로만 조회한다.
+    """
+
+    def __init__(self, auth: AuthService = Depends(AuthService), session_crud: ISessionCRUD = Depends(SessionCRUD)):
+        self.auth = auth
+        self.session_crud = session_crud
+
+    async def list_sessions(self, user_info: UserInfo, access_token: str, status, qt_type, order: str, page: int, page_size: int) -> Res_SessionList:
+        res = Res_SessionList()
+
+        # 1) 인증 (활성 + 저장된 access 토큰 대조)
+        err_type, info = await self.auth.authenticate(user_info, access_token)
+        if err_type != ErrorType.SUCCESS:
+            res.result.SetResult(err_type)
+            return res
+
+        supplier_id = uuid.UUID(info.supplier_id)
+        offset = (page - 1) * page_size
+
+        # 2) 목록 조회 (NEGOTIATION Read 세션, sessions ⨝ items)
+        err_type, rows = await DB_SESSION_MNG.execute_lambda(
+            sessions.DBType(),
+            DBWRType.DB_READ.value,
+            lambda s: self.session_crud.list_by_supplier(s, supplier_id, status, qt_type, order, offset, page_size),
+        )
+        if err_type != ErrorType.SUCCESS:
+            res.result.SetResult(err_type)
+            return res
+
+        # 3) 총개수 (페이지네이션용)
+        err_type, total = await DB_SESSION_MNG.execute_lambda(
+            sessions.DBType(),
+            DBWRType.DB_READ.value,
+            lambda s: self.session_crud.count_by_supplier(s, supplier_id, status, qt_type),
+        )
+        if err_type != ErrorType.SUCCESS:
+            res.result.SetResult(err_type)
+            return res
+
+        res.items = [
+            ListItem(
+                session_id=str(r[0]),
+                session_status=r[1],
+                qt_type=r[2],
+                qt_number=r[3],
+                qt_end_time=r[4].isoformat(timespec="seconds") if r[4] else "",
+                item_code=r[5] or "",
+                item_name=r[6] or "",
+                model_name=r[7] or "",
+                maker_name=r[8] or "",
+            )
+            for r in rows
+        ]
+        res.total = total
+        res.page = page
+        res.page_size = page_size
+        return res
+
+    async def participate(self, user_info: UserInfo, access_token: str, session_id_str: str) -> Res_Participate:
+        res = Res_Participate()
+
+        # 1) 인증
+        err_type, info = await self.auth.authenticate(user_info, access_token)
+        if err_type != ErrorType.SUCCESS:
+            res.result.SetResult(err_type)
+            return res
+        try:
+            session_id = uuid.UUID(session_id_str)
+        except (ValueError, TypeError):
+            res.result.SetResult(ErrorType.NEGO_NOT_FOUND)
+            return res
+
+        # 2) 세션 조회
+        err_type, sess = await DB_SESSION_MNG.execute_lambda(
+            sessions.DBType(),
+            DBWRType.DB_READ.value,
+            lambda s: self.session_crud.get_session_by_id(s, session_id),
+        )
+        if err_type != ErrorType.SUCCESS or sess is None:
+            res.result.SetResult(ErrorType.NEGO_NOT_FOUND)
+            return res
+
+        # 3) 소유 검증 (세션 공급사 == 접속 유저 공급사)
+        if str(sess.supplier_id) != info.supplier_id:
+            res.result.SetResult(ErrorType.NEGO_FORBIDDEN)
+            return res
+
+        # 4) 세션 상태 검증 (미참여/협상거부는 참여 불가)
+        if sess.status in (SessionStatus.NOT_PARTICIPATED.value, SessionStatus.REJECTED.value):
+            res.result.SetResult(ErrorType.NEGO_NOT_PARTICIPABLE)
+            return res
+
+        # 5) 견적 조회 + 마감 상태
+        err_type, quote = await DB_SESSION_MNG.execute_lambda(
+            sessions.DBType(),
+            DBWRType.DB_READ.value,
+            lambda s: self.session_crud.get_quotation_by_id(s, sess.quotation_id),
+        )
+        if err_type != ErrorType.SUCCESS or quote is None:
+            res.result.SetResult(ErrorType.NEGO_NOT_FOUND)
+            return res
+        if quote.status == QuotationStatus.CLOSED.value:
+            res.result.SetResult(ErrorType.NEGO_QUOTATION_CLOSED)
+            return res
+
+        # 6) 마감 시간 초과 (견적 end_time < 현재). 협상생성(1)일 때만 session→미참여로 정리.
+        end = quote.end_time
+        if end is not None and end.tzinfo is None:
+            end = end.replace(tzinfo=timezone.utc)
+        if end is not None and end < datetime.now(timezone.utc):
+            if sess.status == SessionStatus.CREATED.value:
+                await DB_SESSION_MNG.execute_lambda_run(
+                    [sessions.DBType()],
+                    [lambda s: self.session_crud.update_session_status(s, session_id, SessionStatus.NOT_PARTICIPATED.value)],
+                )
+            res.result.SetResult(ErrorType.NEGO_DEADLINE_PASSED)
+            return res
+
+        # 7) 참여 성공 — 협상생성(1)일 때만 상태 전이(협상중/완료는 무변경 진입)
+        if sess.status == SessionStatus.CREATED.value:
+            err_type = await DB_SESSION_MNG.execute_lambda_run(
+                [sessions.DBType()],
+                [
+                    lambda s: self.session_crud.update_session_status(s, session_id, SessionStatus.IN_PROGRESS.value),
+                    lambda s: self.session_crud.update_quotation_status(s, sess.quotation_id, QuotationStatus.IN_PROGRESS.value),
+                ],
+            )
+            if err_type != ErrorType.SUCCESS:
+                res.result.SetResult(err_type)
+                return res
+
+        res.session_id = str(sess.session_id)
+        return res

backend/tests/test_negotiation.py

@@ -0,0 +1,211 @@
+"""협상 도메인 e2e 테스트 (세션 목록 + 참여).
+
+dev negosium_db 를 그대로 쓰므로 전용 테스트 행만 시드/정리한다.
+목록의 qt_end_time 은 quotation.end_time 기준이라 세션마다 견적을 함께 시드한다.
+"""
+
+import uuid
+
+import bcrypt
+import pytest_asyncio
+from sqlalchemy import text
+
+TEST_LOGIN_ID = "pytest_nego_user"
+TEST_PW = "pytest1234"
+TEST_SUPPLIER_NAME = "파이테스트협상공급사"
+MARK = "PYTESTNEGO-"  # 시드 식별용 prefix (item code / qt number)
+
+
+@pytest_asyncio.fixture
+async def nego_seed(db_engine):
+    """공급사 + 유저 + 세션/견적 3건(본인) + 1건(타 공급사) 시드. 세션/견적 id 를 반환."""
+    supplier_id = uuid.uuid4()
+    other_supplier_id = uuid.uuid4()
+    pw_hash = bcrypt.hashpw(TEST_PW.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
+
+    # (code, session.status, qt_type, 마감까지 시간(h), quotation.status, 소속 공급사)
+    specs = [
+        ("A", 1, 2, 2, 1, supplier_id),        # 협상생성 / 재견적 / +2h / 견적생성
+        ("B", 2, 1, 1, 2, supplier_id),        # 협상중   / 재협상 / +1h / 견적진행중
+        ("C", 3, 2, 3, 2, supplier_id),        # 협상완료 / 재견적 / +3h / 견적진행중
+        ("X", 1, 1, 1, 1, other_supplier_id),  # 타 공급사 → 목록/참여에서 제외/차단
+    ]
+    sids, qids = {}, {}
+
+    async def _cleanup(conn):
+        await conn.execute(text(f"DELETE FROM negotiation.sessions WHERE qt_number LIKE '{MARK}%'"))
+        await conn.execute(text(f"DELETE FROM quotation.quotations WHERE number LIKE '{MARK}%'"))
+        await conn.execute(text(f"DELETE FROM partner.items WHERE code LIKE '{MARK}%'"))
+        await conn.execute(text("DELETE FROM supplier.supplier_users WHERE id = :id"), {"id": TEST_LOGIN_ID})
+        await conn.execute(text("DELETE FROM partner.suppliers WHERE name = :n"), {"n": TEST_SUPPLIER_NAME})
+
+    async with db_engine.begin() as conn:
+        await _cleanup(conn)
+        await conn.execute(
+            text("INSERT INTO partner.suppliers (supplier_id, company_id, user_id, name) VALUES (:sid, gen_random_uuid(), gen_random_uuid(), :name)"),
+            {"sid": supplier_id, "name": TEST_SUPPLIER_NAME},
+        )
+        await conn.execute(
+            text(
+                "INSERT INTO supplier.supplier_users (supplier_id, id, password, name, last_accessed_at, status, role) "
+                "VALUES (:sid, :id, :pw, '협상담당자', now(), 1, 1)"
+            ),
+            {"sid": supplier_id, "id": TEST_LOGIN_ID, "pw": pw_hash},
+        )
+        for code, sess_st, qt_type, hrs, quote_st, sup in specs:
+            item_id, qt_id, session_id = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
+            sids[code], qids[code] = session_id, qt_id
+            await conn.execute(
+                text(
+                    "INSERT INTO partner.items (item_id, company_id, user_id, name, code, model_name, manufacturer) "
+                    "VALUES (:iid, gen_random_uuid(), gen_random_uuid(), :name, :code, :model, '테스트제조사')"
+                ),
+                {"iid": item_id, "name": f"상품 {code}", "code": f"{MARK}{code}", "model": f"MODEL-{code}"},
+            )
+            await conn.execute(
+                text(
+                    "INSERT INTO quotation.quotations (qt_id, user_id, qt_setting_id, version_id, name, number, type, status, start_time, end_time) "
+                    "VALUES (:qid, gen_random_uuid(), gen_random_uuid(), gen_random_uuid(), :name, :num, :tp, :st, now(), now() + make_interval(hours => :hrs))"
+                ),
+                {"qid": qt_id, "name": f"견적 {code}", "num": f"{MARK}{code}", "tp": qt_type, "st": quote_st, "hrs": hrs},
+            )
+            await conn.execute(
+                text(
+                    "INSERT INTO negotiation.sessions "
+                    "(session_id, quotation_id, item_id, supplier_id, qt_number, qt_round, qt_type, target_price, status, end_time) "
+                    "VALUES (:sesid, :qid, :iid, :sup, :qtn, 1, :qtt, 100000, :st, now())"
+                ),
+                {"sesid": session_id, "qid": qt_id, "iid": item_id, "sup": sup, "qtn": f"{MARK}{code}", "qtt": qt_type, "st": sess_st},
+            )
+
+    yield {"supplier_id": supplier_id, "sids": sids, "qids": qids}
+
+    async with db_engine.begin() as conn:
+        await _cleanup(conn)
+
+
+async def _login_token(client):
+    r = await client.post("/v1/auth/login", json={"id": TEST_LOGIN_ID, "pw": TEST_PW})
+    return r.json()["access_token"]
+
+
+async def _list(client, token, **params):
+    return await client.get("/v1/negotiation/sessions", headers={"Authorization": f"Bearer {token}"}, params=params)
+
+
+async def _participate(client, token, session_id):
+    return await client.post(f"/v1/negotiation/sessions/{session_id}/participate", headers={"Authorization": f"Bearer {token}"})
+
+
+async def _session_status(db_engine, session_id):
+    async with db_engine.begin() as conn:
+        return (await conn.execute(text("SELECT status FROM negotiation.sessions WHERE session_id = :sid"), {"sid": session_id})).scalar()
+
+
+async def _quotation_status(db_engine, qt_id):
+    async with db_engine.begin() as conn:
+        return (await conn.execute(text("SELECT status FROM quotation.quotations WHERE qt_id = :qid"), {"qid": qt_id})).scalar()
+
+
+# ---- 목록 -------------------------------------------------------------------
+async def test_list_returns_only_own_supplier_sessions(client, nego_seed):
+    token = await _login_token(client)
+    body = (await _list(client, token)).json()
+    assert body["result"]["success"] is True
+    assert body["total"] == 3  # 본인 공급사 3건만 (타 공급사 X 제외)
+    one = next(i for i in body["items"] if i["item_code"] == f"{MARK}B")
+    assert one["session_status"] == 2 and one["qt_type"] == 1
+    assert one["model_name"] == "MODEL-B" and one["maker_name"] == "테스트제조사"
+    assert one["session_id"] and one["qt_end_time"]
+
+
+async def test_list_filter_status(client, nego_seed):
+    token = await _login_token(client)
+    body = (await _list(client, token, status=2)).json()
+    assert body["total"] == 1 and body["items"][0]["item_code"] == f"{MARK}B"
+
+
+async def test_list_filter_qt_type(client, nego_seed):
+    token = await _login_token(client)
+    body = (await _list(client, token, qt_type=2)).json()
+    assert {i["item_code"] for i in body["items"]} == {f"{MARK}A", f"{MARK}C"}
+
+
+async def test_list_order_by_quotation_end_time(client, nego_seed):
+    token = await _login_token(client)
+    asc = (await _list(client, token, order="asc")).json()["items"]
+    desc = (await _list(client, token, order="desc")).json()["items"]
+    assert asc[0]["item_code"] == f"{MARK}B"   # +1h 가 가장 임박
+    assert desc[0]["item_code"] == f"{MARK}C"  # +3h 가 가장 멈
+
+
+async def test_list_pagination(client, nego_seed):
+    token = await _login_token(client)
+    body = (await _list(client, token, page=1, page_size=2)).json()
+    assert body["total"] == 3 and len(body["items"]) == 2
+
+
+async def test_list_requires_auth(client):
+    assert (await client.get("/v1/negotiation/sessions")).status_code in (401, 403)
+
+
+# ---- 참여 -------------------------------------------------------------------
+async def test_participate_success(client, nego_seed, db_engine):
+    token = await _login_token(client)
+    sid, qid = nego_seed["sids"]["A"], nego_seed["qids"]["A"]  # 협상생성
+    r = await _participate(client, token, sid)
+    assert r.json()["result"]["success"] is True
+    assert r.json()["session_id"] == str(sid)
+    assert await _session_status(db_engine, sid) == 2      # 협상중
+    assert await _quotation_status(db_engine, qid) == 2    # 견적진행중
+
+
+async def test_participate_forbidden_other_supplier(client, nego_seed):
+    token = await _login_token(client)
+    r = await _participate(client, token, nego_seed["sids"]["X"])  # 타 공급사 세션
+    assert r.json()["result"]["code"] == 1300  # NEGO_FORBIDDEN
+
+
+async def test_participate_not_participable(client, nego_seed, db_engine):
+    token = await _login_token(client)
+    sid = nego_seed["sids"]["A"]
+    async with db_engine.begin() as conn:
+        await conn.execute(text("UPDATE negotiation.sessions SET status = 4 WHERE session_id = :sid"), {"sid": sid})  # 미참여
+    r = await _participate(client, token, sid)
+    assert r.json()["result"]["code"] == 1301  # NEGO_NOT_PARTICIPABLE
+
+
+async def test_participate_quotation_closed(client, nego_seed, db_engine):
+    token = await _login_token(client)
+    sid, qid = nego_seed["sids"]["A"], nego_seed["qids"]["A"]
+    async with db_engine.begin() as conn:
+        await conn.execute(text("UPDATE quotation.quotations SET status = 3 WHERE qt_id = :qid"), {"qid": qid})  # 견적마감
+    r = await _participate(client, token, sid)
+    assert r.json()["result"]["code"] == 1302  # NEGO_QUOTATION_CLOSED
+
+
+async def test_participate_deadline_passed_sets_not_participated(client, nego_seed, db_engine):
+    token = await _login_token(client)
+    sid, qid = nego_seed["sids"]["A"], nego_seed["qids"]["A"]  # 협상생성
+    async with db_engine.begin() as conn:
+        await conn.execute(text("UPDATE quotation.quotations SET end_time = now() - make_interval(hours => 1) WHERE qt_id = :qid"), {"qid": qid})
+    r = await _participate(client, token, sid)
+    assert r.json()["result"]["code"] == 1303         # NEGO_DEADLINE_PASSED
+    assert await _session_status(db_engine, sid) == 4  # 협상생성이었으므로 미참여로 정리됨
+    assert await _quotation_status(db_engine, qid) == 1  # 견적은 변경 안 됨
+
+
+async def test_participate_in_progress_no_state_change(client, nego_seed, db_engine):
+    token = await _login_token(client)
+    sid, qid = nego_seed["sids"]["B"], nego_seed["qids"]["B"]  # 이미 협상중
+    r = await _participate(client, token, sid)
+    assert r.json()["result"]["success"] is True
+    assert r.json()["session_id"] == str(sid)
+    assert await _session_status(db_engine, sid) == 2      # 무변경 (협상중 유지)
+    assert await _quotation_status(db_engine, qid) == 2    # 무변경
+
+
+async def test_participate_session_not_found(client, nego_seed):
+    token = await _login_token(client)
+    r = await _participate(client, token, str(uuid.uuid4()))
+    assert r.json()["result"]["code"] == 1304  # NEGO_NOT_FOUND