[fix] 견적 자동마감 동시성·정합성 + 프론트 안정화 (코드리뷰 후속)

백엔드 close_and_decide 경로:
- 동시 이중 마감 가드: 마감 판정 전 원자적 CLOSED 선점(claim)으로 두 크론 잡·수동마감 경합 직렬화
- 재생성 실패 표면화: regenerate_next_round 결과 검사 → 실패 시 REGEN_FAILED 반환(체인 끊김 은폐 방지)
- 차수 충돌 방지: 다음 라운드 = 체인 최신 round+1(chain_max_round 기준)
- 재생성 사유 집계 정밀화: 미참여/동가를 양성 표식으로 구분(단독낙찰·거부 오집계 제거)
- 재생성 라운드 최소 협상기간 하한(즉시 재마감 캐스케이드 방지)
- 잡 루프 per-item 예외 격리(한 건 실패가 배치 전체를 멈추지 않음)

프론트:
- useChatController: 무권한 가드를 sessionId 별로 추적해 세션 변경 시 자연 해제
- useScrollLock: 마지막 해제를 rAF 로 지연해 재마운트 사이 일시적 잠금 해제 방지
- quotation 상세 쿼리 placeholderData 로 라운드 전환 중 시트 유지

테스트:
- 신규 test_close_and_decide_fixes.py(동시성·차수·집계·기간 하한 검증)
- conftest 결함 수정(존재하지 않는 tbl_account TRUNCATE 제거, companies.status 명시)
- stale 테스트 갱신(test_quotation_create 를 타입드 Req/신규 응답 형식에 맞게 재작성)

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

frontend/src/features/chat/hooks/useChatController.ts

@@ -71,13 +71,15 @@ export function useChatController(sessionId: string) {
   // 진입 로드 실패. 권한 없음/없는 세션(잘못된 접근)이면 토스트 후 목록으로 복귀시킨다.
   const loadError = initQuery.error ?? messagesQuery.error
   const isInvalidAccess = isApiError(loadError) && INVALID_ACCESS_CODES.has(loadError.code)
-  const redirectedRef = useRef(false)
+  // 이미 리다이렉트한 sessionId 를 기록(boolean 이 아니라 sessionId). 리마운트 없이 sessionId 가 바뀌면
+  // (브라우저 뒤로/앞으로 등) 값이 달라져 가드가 자연 해제 → 두 번째 무권한 세션도 토스트+복귀가 동작한다.
+  const redirectedSessionRef = useRef<string | null>(null)
   useEffect(() => {
-    if (!isInvalidAccess || redirectedRef.current) return
+    if (!isInvalidAccess || redirectedSessionRef.current === sessionId) return
-    redirectedRef.current = true
+    redirectedSessionRef.current = sessionId
     toast.error('잘못된 접근입니다.')
     navigate('/list', { replace: true })
-  }, [isInvalidAccess, navigate])
+  }, [isInvalidAccess, sessionId, navigate])
 
   // init 메타 → 스토어
   useEffect(() => {

negodata/backend/common/database/db_session_manager.py

@@ -156,6 +156,25 @@ class DBSessionManager(Singleton):
                 raise RuntimeError(err_type.name, err_msg)
             return err_type
 
+    async def add_with_rowcount(self, db: AsyncSession, query, err_msg="DB Operation Failed") -> tuple[ErrorType, int]:
+        """update/delete 등 비-select 쿼리 실행 후 (ErrorType, 영향행수) 반환.
+        조건부 갱신(WHERE 로 상태를 거른 UPDATE)이 실제로 적용됐는지 판별하는 동시처리 가드용."""
+        try:
+            if hasattr(query, "column_descriptions"):
+                raise RuntimeError("DO NOT USE SELECT QUERY IN DBJOB")
+            res = await db.execute(query, execution_options=immutabledict({"synchronize_session": "fetch"}))
+            return ErrorType.SUCCESS, res.rowcount
+        except IntegrityError as ex:
+            await db.rollback()
+            err_type = ErrorType.DB_ALREADY_SAME_KEY
+            LOG.e_no_callstack(f"[{err_type.name}] {err_msg=}, {ex=}")
+            return err_type, 0
+        except Exception as ex:
+            await db.rollback()
+            err_type = ErrorType.DB_RUN_FAILED
+            LOG.e_no_callstack(f"[{err_type.name}] {err_msg=}, {ex=}")
+            return err_type, 0
+
     async def execute(self, db: AsyncSession, query, err_msg="DB Query Execution Failed", raise_error=True) -> tuple[ErrorType, list]:
         """select 쿼리 실행 후 결과 리스트 반환."""
         try:
@@ -201,5 +220,22 @@ class DBSessionManager(Singleton):
         finally:
             await self.end_session(db_type, DBWRType.DB_WRITE.value)
 
+    async def execute_lambda_claim(self, db_type: int, func) -> tuple[ErrorType, int]:
+        """조건부 변경 쿼리 1건을 한 트랜잭션으로 실행/commit 하고 (ErrorType, 적용행수) 반환.
+        동시처리 가드용 — func(session) -> (ErrorType, rowcount). 적용행수 0 이면 다른 호출자가 이미 처리한 것.
+        (Postgres READ COMMITTED 에서 같은 행 UPDATE 는 행 잠금으로 직렬화되어, 진 호출자는 0 을 받는다.)"""
+        s = await self.start_session(db_type, DBWRType.DB_WRITE.value)
+        try:
+            err_type, rowcount = await func(s)
+            if err_type != ErrorType.SUCCESS:
+                return err_type, 0
+            commit_err = await self.run(s)
+            return commit_err, rowcount
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED, 0
+        finally:
+            await self.end_session(db_type, DBWRType.DB_WRITE.value)
+
 
 DB_SESSION_MNG = DBSessionManager()

negodata/backend/common/enums.py

@@ -148,7 +148,8 @@ class CloseOutcome(Enum):
 
     AWARDED = "awarded"          # 단독 낙찰 확정
     REGENERATED = "regenerated"  # 다음 라운드 재생성
-    CLOSED = "closed"            # 그냥 마감
+    CLOSED = "closed"            # 그냥 마감 (선점 실패로 이미 닫혀 있던 경우 포함)
+    REGEN_FAILED = "regen_failed"  # 재생성 시도했으나 실패 — 원본은 CLOSED 인데 다음 라운드가 없음(체인 끊김, 모니터링 필요)
 
 
 class ChatSender(CodeEnum):

negodata/backend/conftest.py

@@ -13,6 +13,7 @@ from sqlalchemy import text
 from sqlalchemy.ext.asyncio import create_async_engine
 
 from common.database.model.models import MAIN_BASE
+from common.enums import CompanyStatus
 from config.server_configs import main_db_config
 
 
@@ -50,7 +51,7 @@ async def db_engine():
         # negodata 도메인 테이블 전부 비워 격리 (CASCADE: FK 미설정이라 안전망)
         await conn.execute(
             text(
-                "TRUNCATE TABLE tbl_account, users, companies, items, suppliers, "
+                "TRUNCATE TABLE users, companies, items, suppliers, "
                 "quotation_settings, quotations, sessions RESTART IDENTITY CASCADE"
             )
         )
@@ -65,9 +66,10 @@ async def company_id(db_engine) -> str:
     """
     cid = uuid.uuid4()
     async with db_engine.begin() as conn:
+        # status 는 NOT NULL(모델 default 는 ORM 전용이라 raw INSERT 엔 안 먹음) → 명시.
         await conn.execute(
-            text("INSERT INTO companies (company_id, name) VALUES (:cid, :name)"),
+            text("INSERT INTO companies (company_id, name, status) VALUES (:cid, :name, :status)"),
-            {"cid": cid, "name": "테스트사"},
+            {"cid": cid, "name": "테스트사", "status": CompanyStatus.ACTIVE.value},
         )
     return str(cid)
 

negodata/backend/crud/quotation_crud.py

@@ -97,7 +97,7 @@ class IQuotationCRUD(ABC):
         pass
 
     @abstractmethod
-    async def list_chain_equal_flags(self, cdb: AsyncSession, number, current_round) -> Tuple[ErrorType, list]:
+    async def list_chain_close_flags(self, cdb: AsyncSession, number, current_round) -> Tuple[ErrorType, list]:
         pass
 
     @abstractmethod
@@ -120,6 +120,10 @@ class IQuotationCRUD(ABC):
     async def bulk_update_sessions_status(self, cdb: AsyncSession, qt_ids, from_statuses: list[int], to_status: int) -> ErrorType:
         pass
 
+    @abstractmethod
+    async def claim_for_close(self, cdb: AsyncSession, qt_id) -> Tuple[ErrorType, int]:
+        pass
+
 
 class QuotationCRUD(IQuotationCRUD):
     async def search(
@@ -348,6 +352,26 @@ class QuotationCRUD(IQuotationCRUD):
             LOG.e_no_callstack(ex)
             return ErrorType.DB_RUN_FAILED
 
+    async def claim_for_close(self, cdb: AsyncSession, qt_id) -> Tuple[ErrorType, int]:
+        """[동시 마감 가드] 아직 안 닫힌(status != CLOSED, not deleted) 견적만 CLOSED 로 선점 전이.
+        반환: (ErrorType, 적용행수). 동시 호출 시 Postgres 행 잠금으로 직렬화되어
+        실제로 CLOSED 로 바꾼 호출자만 1, 이미 닫혀 있던(진 호출자/재처리) 경우는 0 을 받는다.
+        close_and_decide 가 이 결과로 '마감 판정 권한'을 단 한 번만 갖도록 한다."""
+        try:
+            query = (
+                update(quotations)
+                .where(
+                    quotations.qt_id == qt_id,
+                    quotations.status != QuotationStatus.CLOSED.value,
+                    quotations.deleted == False,  # noqa: E712
+                )
+                .values(status=QuotationStatus.CLOSED.value, updated_at=GTime.UTC())
+            )
+            return await DB_SESSION_MNG.add_with_rowcount(cdb, query)
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED, 0
+
     async def update_sessions_status(self, cdb: AsyncSession, qt_id, from_statuses: list[int], to_status: int) -> ErrorType:
         # 견적에 딸린 세션 중 from_statuses 에 속한 것만 to_status 로 일괄 전이(삭제 제외). 다른 상태는 건드리지 않는다.
         try:
@@ -447,11 +471,15 @@ class QuotationCRUD(IQuotationCRUD):
             LOG.e_no_callstack(ex)
             return ErrorType.DB_RUN_FAILED, []
 
-    async def list_chain_equal_flags(self, cdb: AsyncSession, number, current_round) -> Tuple[ErrorType, list]:
+    async def list_chain_close_flags(self, cdb: AsyncSession, number, current_round) -> Tuple[ErrorType, list]:
-        """[재생성 한도] 같은 견적번호(체인)의 이전 라운드(round < current_round)들의 equal_bid_yn 목록. 삭제 제외.
+        """[재생성 한도] 같은 견적번호(체인)의 이전 라운드(round < current_round)들의 (preferred_sp_yn, equal_bid_yn) 목록. 삭제 제외.
-        True=동가로 닫힌 라운드 / 그 외(False·NULL)=미참여로 닫힌 라운드."""
+        마감 사유 식별용 표식:
+          - equal_bid_yn=True                         → 동가 재생성
+          - preferred_sp_yn=False AND equal_bid_yn=False → 미참여 재생성
+          - preferred_sp_yn=True                       → 단독낙찰(체인 어느 쪽에도 안 셈)
+          - 둘 다 NULL                                  → 거부/한도 그냥 마감(안 셈)"""
         try:
-            query = select(quotations.equal_bid_yn).where(
+            query = select(quotations.preferred_sp_yn, quotations.equal_bid_yn).where(
                 quotations.number == number,
                 quotations.round < current_round,
                 quotations.deleted == False,  # noqa: E712

negodata/backend/scheduler/jobs.py

@@ -15,6 +15,27 @@ from crud.quotation_crud import QuotationCRUD
 from services.quotation_service import QuotationService
 
 
+async def _close_each(service: QuotationService, qt_ids) -> Counter:
+    """대상 견적마다 close_and_decide 를 호출하되, 한 건의 예외가 배치 전체를 멈추지 않도록 격리한다.
+    (예전 per-item try/continue 보존 — 한 견적의 DB 오류 등으로 나머지 견적이 이번 tick 에서 누락되면 안 됨.)
+    반환: 결과(CloseOutcome) 카운트 + 예외 발생 건수('error')."""
+    results = Counter()
+    for qt_id in qt_ids:
+        try:
+            results[await service.close_and_decide(qt_id)] += 1
+        except Exception as ex:
+            results["error"] += 1
+            LOG.e_no_callstack(f"[scheduler] close_and_decide 실패 qt={qt_id}: {ex}")
+    return results
+
+
+def _format_results(results: Counter) -> str:
+    return (
+        f"낙찰 {results[CloseOutcome.AWARDED]} / 재생성 {results[CloseOutcome.REGENERATED]} / "
+        f"재생성실패 {results[CloseOutcome.REGEN_FAILED]} / 마감 {results[CloseOutcome.CLOSED]} / 오류 {results['error']}"
+    )
+
+
 async def close_expired_quotations() -> int:
     """[잡①] 마감일이 지난 견적을 자동 마감 처리한다. 하루 한 번 실행.
     대상: 마감 시각이 이미 지났는데 아직 마감되지 않은(삭제되지도 않은) 견적.
@@ -32,11 +53,9 @@ async def close_expired_quotations() -> int:
         LOG.e_no_callstack(f"[scheduler] close_expired 대상 조회 실패: {err_type.name}")
         return 0
 
-    results = Counter()
+    results = await _close_each(service, qt_ids)
-    for qt_id in qt_ids:
-        results[await service.close_and_decide(qt_id)] += 1
     if results:
-        LOG.i(f"[scheduler] close_expired: 낙찰 {results[CloseOutcome.AWARDED]} / 재생성 {results[CloseOutcome.REGENERATED]} / 마감 {results[CloseOutcome.CLOSED]}")
+        LOG.i(f"[scheduler] close_expired: {_format_results(results)}")
     return sum(results.values())
 
 
@@ -56,9 +75,7 @@ async def close_negotiated_quotations() -> int:
         LOG.e_no_callstack(f"[scheduler] close_negotiated 대상 조회 실패: {err_type.name}")
         return 0
 
-    results = Counter()
+    results = await _close_each(service, qt_ids)
-    for qt_id in qt_ids:
-        results[await service.close_and_decide(qt_id)] += 1
     if results:
-        LOG.i(f"[scheduler] close_negotiated: 낙찰 {results[CloseOutcome.AWARDED]} / 재생성 {results[CloseOutcome.REGENERATED]} / 마감 {results[CloseOutcome.CLOSED]}")
+        LOG.i(f"[scheduler] close_negotiated: {_format_results(results)}")
     return sum(results.values())

negodata/backend/services/quotation_service.py

@@ -1,6 +1,6 @@
 import re
 import uuid
-from datetime import timezone
+from datetime import timezone, timedelta
 from typing import Optional
 
 from fastapi import Depends
@@ -8,6 +8,7 @@ from fastapi import Depends
 from common.database.db_session_manager import DB_SESSION_MNG
 from common.database.model.models import quotations, sessions, chats, versions, version_nego_cards, version_wild_cards
 from common.enums import CloseOutcome, DBWRType, ErrorType, QuotationStatus, QuotationType, SessionStatus
+from common.logger import LOG
 from common.models.gmodel import PageParams
 from common.utils.gtime import GTime
 from config.server_configs import web_server_config
@@ -43,6 +44,12 @@ class QuotationService:
     # 재생성 한도: 한 체인(같은 견적번호)에서 사유(미참여/동가)별 최대 1번까지 재생성(순서 무관, 같은 사유 2번 불가).
     MAX_REGEN_PER_CAUSE = 1
 
+    # 재생성 라운드의 최소 협상기간(방어적 하한). 원본 협상기간이 비정상적으로 짧으면(또는 0/음수면)
+    # 새 라운드가 생성 즉시 만료돼 다음 크론 tick(*/5분)에 또 마감되는 연쇄를 막는다.
+    # 정상 견적(수 시간~수일)은 원본 기간을 그대로 쓰며, 이 하한은 비정상적으로 짧은 경우에만 적용된다.
+    # TODO 하한값 변경 해야함 !!! feat. MarineYang
+    MIN_REGEN_DURATION = timedelta(hours=1)
+
     def __init__(self, quotation_crud: IQuotationCRUD = Depends(QuotationCRUD)):
         self.quotation_crud = quotation_crud
 
@@ -190,9 +197,18 @@ class QuotationService:
 
         # 3) 다음 라운드의 견적 생성
         now = GTime.UTC()
-        duration = original.end_time - original.start_time
+        # 원본 협상기간을 이어쓰되, 비정상적으로 짧으면 최소 하한을 적용(즉시 만료→연쇄 재마감 방지).
-        # 진입 경로(크론 마감 / 수동 regenerate_quotation) 모두 '마지막 차수'만 넘기므로 +1 이 곧 체인 다음 차수.
+        duration = max(original.end_time - original.start_time, self.MIN_REGEN_DURATION)
-        next_round = original.round + 1
+        # 다음 차수는 '원본 round+1' 이 아니라 '체인(같은 번호) 최신 round+1'.
+        # 크론 마감과 수동 regenerate_quotation 이 같은 체인을 처리하는 타이밍이 엇갈려도
+        # 항상 체인 끝에 이어붙어 uq_quotations_number(number, round) 충돌을 막는다.
+        _e, chain_max = await DB_SESSION_MNG.execute_lambda(
+            quotations.DBType(),
+            DBWRType.DB_READ.value,
+            lambda s: self.quotation_crud.chain_max_round(s, original.number),
+        )
+        base_round = chain_max if (_e == ErrorType.SUCCESS and chain_max) else original.round
+        next_round = base_round + 1
         # 이름에 '(N차)' 표기. 원래 이름 기준(기존 '(M차)' 표기는 떼고 새로) + name 컬럼 50자 제한 보호.
         suffix = f" ({next_round}차)"
         base_name = re.sub(r"\s*\(\d+차\)\s*$", "", original.name or "")[: 50 - len(suffix)]
@@ -372,6 +388,26 @@ class QuotationService:
             ],
         )
 
+    async def _close_as_no_show(self, qt_uuid) -> None:
+        """전원 미참여로 '다음 라운드 재생성' 하며 마감 + 미완료 세션 미참여.
+        재생성 사유(미참여)를 체인에 남기기 위해 preferred_sp_yn=False, equal_bid_yn=False 로 양성 표식한다
+        (단독낙찰=preferred_sp_yn True / 동가=equal_bid_yn True / 거부·한도 등 그냥 마감=둘 다 NULL 과 구분).
+        _chain_regen_counts 가 이 표식으로 '미참여 재생성 이력'만 정확히 센다."""
+        data = {
+            "status": QuotationStatus.CLOSED.value,
+            "preferred_sp_yn": False,
+            "equal_bid_yn": False,
+        }
+        await DB_SESSION_MNG.execute_lambda_run(
+            [quotations.DBType()],
+            [
+                lambda s: self.quotation_crud.update_quotation(s, qt_uuid, data),
+                lambda s: self.quotation_crud.update_sessions_status(
+                    s, qt_uuid, [SessionStatus.CREATED.value, SessionStatus.IN_PROGRESS.value], SessionStatus.NOT_PARTICIPATED.value
+                ),
+            ],
+        )
+
     async def _close_as_equal(self, qt_uuid, equal) -> None:
         """동가로 마감 + 미완료 세션 미참여. equal_bid_yn/data 를 기록해 둔다
         (재생성 한도 계산이 이 플래그로 동가 라운드를 식별하고, 프론트도 동가 정보를 그대로 쓴다)."""
@@ -407,6 +443,17 @@ class QuotationService:
         if err_type != ErrorType.SUCCESS or original is None:
             return CloseOutcome.CLOSED
 
+        # [동시 마감 가드] 마감 판정 전에 원자적으로 status→CLOSED 를 선점한다.
+        # 두 크론 잡(close_expired / close_negotiated)이나 수동 stop_quotation 이 같은 견적을
+        # 동시에 닫으려 해도, 실제로 CLOSED 로 전이한 호출자만 통과하고 진 호출자는 여기서 끝난다
+        # → 이중 재생성·uq(number,round) 충돌 방지. (이미 닫힌 견적의 재처리도 여기서 차단)
+        claim_err, claimed = await DB_SESSION_MNG.execute_lambda_claim(
+            quotations.DBType(),
+            lambda s: self.quotation_crud.claim_for_close(s, qt_uuid),
+        )
+        if claim_err != ErrorType.SUCCESS or claimed == 0:
+            return CloseOutcome.CLOSED
+
         err_type, rows = await DB_SESSION_MNG.execute_lambda(
             sessions.DBType(),
             DBWRType.DB_READ.value,
@@ -430,7 +477,14 @@ class QuotationService:
         if equal is not None and equal_used < self.MAX_REGEN_PER_CAUSE:
             tied_ids = [uuid.UUID(sp["supplier_id"]) for sp in equal["suppliers"]]
             await self._close_as_equal(qt_uuid, equal)  # 동가 기록(equal_bid_yn) 후 마감
-            await self.regenerate_next_round(qt_uuid, tied_ids)
+            regen = await self.regenerate_next_round(qt_uuid, tied_ids)
+            if not regen.result.success:
+                # 원본은 이미 CLOSED 인데 다음 라운드 생성이 실패 → 체인이 끊긴 상태. 성공으로 위장하지 않고 드러낸다.
+                LOG.e_no_callstack(
+                    f"[close] 동가 재생성 실패 qt={qt_uuid} number={original.number} round={original.round} "
+                    f"code={regen.result.code}({regen.result.desc})"
+                )
+                return CloseOutcome.REGEN_FAILED
             return CloseOutcome.REGENERATED
         # 3) 협상거부 있음 → 마감만 (재생성 안 함)
         if has_rejected:
@@ -439,26 +493,36 @@ class QuotationService:
         # 4) 전원 미참여 → 공급사 전체로 다음 라운드 (체인에 미참여 재생성 이력 없을 때만)
         if not done and rows and no_part_used < self.MAX_REGEN_PER_CAUSE:
             supplier_ids = list({r.supplier_id for r in rows})
-            await self._just_close(qt_uuid)
+            await self._close_as_no_show(qt_uuid)  # 미참여 재생성 표식(preferred_sp_yn=False, equal_bid_yn=False) 후 마감
-            await self.regenerate_next_round(qt_uuid, supplier_ids)
+            regen = await self.regenerate_next_round(qt_uuid, supplier_ids)
+            if not regen.result.success:
+                # 원본은 이미 CLOSED 인데 다음 라운드 생성이 실패 → 체인이 끊긴 상태. 성공으로 위장하지 않고 드러낸다.
+                LOG.e_no_callstack(
+                    f"[close] 미참여 재생성 실패 qt={qt_uuid} number={original.number} round={original.round} "
+                    f"code={regen.result.code}({regen.result.desc})"
+                )
+                return CloseOutcome.REGEN_FAILED
             return CloseOutcome.REGENERATED
         # 5) 그 외 / 한도 도달 → 마감만
         await self._just_close(qt_uuid)
         return CloseOutcome.CLOSED
 
     async def _chain_regen_counts(self, number: str, current_round: int) -> tuple[int, int]:
-        """체인(같은 견적번호) 이전 라운드들의 재생성 사유 횟수. 반환: (미참여 횟수, 동가 횟수).
+        """체인(같은 견적번호) 이전 라운드들의 '재생성 사유' 횟수. 반환: (미참여 횟수, 동가 횟수).
-        동가로 닫힌 라운드는 equal_bid_yn=True 로 기록되므로 그 플래그로 센다.
+        마감 시 남긴 양성 표식으로만 센다(오집계 방지):
-        (이전 라운드는 동가 아니면 미참여 둘뿐 — 단독낙찰·거부는 체인을 끝냄 → equal_bid_yn 이 True 아니면 미참여)."""
+          - 동가 재생성    → equal_bid_yn=True
+          - 미참여 재생성  → preferred_sp_yn=False AND equal_bid_yn=False
+        단독낙찰(preferred_sp_yn=True)·거부/한도 그냥 마감(둘 다 NULL)은 어느 쪽에도 세지 않는다.
+        (수동 regenerate_quotation 으로 단독낙찰·거부 라운드를 이어붙여도 자동 재생성 한도에 영향 없음.)"""
         err_type, flags = await DB_SESSION_MNG.execute_lambda(
             quotations.DBType(),
             DBWRType.DB_READ.value,
-            lambda s: self.quotation_crud.list_chain_equal_flags(s, number, current_round),
+            lambda s: self.quotation_crud.list_chain_close_flags(s, number, current_round),
         )
         if err_type != ErrorType.SUCCESS:
             return 0, 0
-        equal = sum(1 for f in flags if f is True)
+        equal = sum(1 for _pref, eq in flags if eq is True)
-        no_part = len(flags) - equal
+        no_part = sum(1 for pref, eq in flags if pref is False and eq is False)
         return no_part, equal
 
     async def regenerate_quotation(self, qt_id: str, supplier_ids: list) -> Res_CreateQuotation:

negodata/backend/tests/test_close_and_decide_fixes.py

@@ -0,0 +1,178 @@
+"""close_and_decide 동시성·정합성 수정 검증 (코드리뷰 후속).
+
+검증 대상:
+  - #2 동시 이중 마감 가드: 같은 견적을 동시에 close_and_decide 해도 다음 라운드는 1개만 생성
+  - #3 차수 매김: 다음 라운드 round = 체인 최신 round + 1
+  - #4 재생성 사유 집계: 단독낙찰(preferred_sp_yn=True) 이전 라운드를 '미참여'로 오집계하지 않음
+  - #6 재생성 라운드 최소 협상기간 하한(즉시 재마감 캐스케이드 방지)
+
+실행 전제: tests/test_scheduler.py 와 동일(PostgreSQL, APP_ENV=test).
+"""
+import asyncio
+import uuid
+from datetime import datetime, timedelta
+
+import pytest_asyncio
+from sqlalchemy import text
+
+from common.enums import CloseOutcome, QuotationStatus, QuotationType, SessionStatus
+from crud.quotation_crud import QuotationCRUD
+from services.quotation_service import QuotationService
+
+PAST = datetime(2020, 1, 1)
+
+
+@pytest_asyncio.fixture
+async def clean(db_engine):
+    async with db_engine.begin() as conn:
+        await conn.execute(text("TRUNCATE TABLE sessions, quotations RESTART IDENTITY CASCADE"))
+    return db_engine
+
+
+async def _seed_quotation(
+    engine, *, number, round_, status, start_time=PAST, end_time=PAST,
+    preferred_sp_yn=None, equal_bid_yn=None,
+):
+    qt_id = uuid.uuid4()
+    async with engine.begin() as conn:
+        await conn.execute(
+            text(
+                "INSERT INTO quotations "
+                "(qt_id, user_id, qt_setting_id, version_id, name, number, type, status, "
+                " round, iteration, start_time, end_time, deleted, preferred_sp_yn, equal_bid_yn) VALUES "
+                "(:qt_id, :user_id, :qt_setting_id, :version_id, :name, :number, :type, :status, "
+                " :round, 0, :start_time, :end_time, false, :pref, :eq)"
+            ),
+            {
+                "qt_id": qt_id, "user_id": uuid.uuid4(), "qt_setting_id": uuid.uuid4(),
+                "version_id": uuid.uuid4(), "name": "견적", "number": number,
+                "type": QuotationType.REQUOTE.value, "status": status, "round": round_,
+                "start_time": start_time, "end_time": end_time,
+                "pref": preferred_sp_yn, "eq": equal_bid_yn,
+            },
+        )
+    return qt_id
+
+
+async def _add_session(engine, qt_id, *, status, bid_price=None, supplier_id=None):
+    async with engine.begin() as conn:
+        await conn.execute(
+            text(
+                "INSERT INTO sessions "
+                "(session_id, quotation_id, item_id, supplier_id, qt_number, qt_round, qt_type, "
+                " target_price, status, bid_price, end_time) VALUES "
+                "(:session_id, :quotation_id, :item_id, :supplier_id, :qt_number, :qt_round, :qt_type, "
+                " 0, :status, :bid_price, :end_time)"
+            ),
+            {
+                "session_id": uuid.uuid4(), "quotation_id": qt_id, "item_id": uuid.uuid4(),
+                "supplier_id": supplier_id or uuid.uuid4(), "qt_number": "Q", "qt_round": 1,
+                "qt_type": QuotationType.REQUOTE.value, "status": status,
+                "bid_price": bid_price, "end_time": PAST,
+            },
+        )
+
+
+async def _rounds(engine, number):
+    """체인(number)의 (round, status, end_time, start_time) 목록 — round 오름차순."""
+    async with engine.begin() as conn:
+        return (await conn.execute(
+            text("SELECT round, status, start_time, end_time FROM quotations "
+                 "WHERE number = :n ORDER BY round"),
+            {"n": number},
+        )).all()
+
+
+# ----- #2 동시 이중 마감 가드 -----
+async def test_concurrent_close_creates_only_one_next_round(clean):
+    """같은 견적을 5번 동시에 close_and_decide 해도 다음 라운드는 정확히 1개만 생성된다."""
+    engine = clean
+    number = "C-CONCURRENT"
+    qt = await _seed_quotation(engine, number=number, round_=1, status=QuotationStatus.ACTIVE.value)
+    # 전원 미참여(미시작 세션만) → close_and_decide 가 '다음 라운드 재생성' 경로를 탄다
+    await _add_session(engine, qt, status=SessionStatus.CREATED.value)
+    await _add_session(engine, qt, status=SessionStatus.CREATED.value)
+
+    service = QuotationService(QuotationCRUD())
+    outcomes = await asyncio.gather(*[service.close_and_decide(qt) for _ in range(5)])
+
+    regenerated = sum(1 for o in outcomes if o == CloseOutcome.REGENERATED)
+    rounds = await _rounds(engine, number)
+    round_numbers = [r.round for r in rounds]
+
+    assert regenerated == 1, f"재생성은 1번만 일어나야 함, 실제 {regenerated} ({outcomes})"
+    assert round_numbers == [1, 2], f"체인은 [1,2] 여야 함(중복/충돌 없음), 실제 {round_numbers}"
+
+
+# ----- #3 차수 + #6 최소 협상기간 하한 -----
+async def test_next_round_numbering_and_min_duration(clean):
+    """다음 라운드 round = 최신+1, 협상기간이 0이어도 최소 하한(MIN_REGEN_DURATION)이 적용된다."""
+    engine = clean
+    number = "C-DURATION"
+    # start==end (협상기간 0) → 하한이 적용되지 않으면 새 라운드도 0 길이가 된다
+    qt = await _seed_quotation(
+        engine, number=number, round_=1, status=QuotationStatus.ACTIVE.value,
+        start_time=PAST, end_time=PAST,
+    )
+    await _add_session(engine, qt, status=SessionStatus.CREATED.value)
+
+    service = QuotationService(QuotationCRUD())
+    outcome = await service.close_and_decide(qt)
+
+    assert outcome == CloseOutcome.REGENERATED
+    rounds = await _rounds(engine, number)
+    assert [r.round for r in rounds] == [1, 2]
+    nxt = rounds[1]
+    duration = nxt.end_time - nxt.start_time
+    assert duration >= QuotationService.MIN_REGEN_DURATION, (
+        f"재생성 라운드 협상기간({duration})이 최소 하한({QuotationService.MIN_REGEN_DURATION}) 이상이어야 함"
+    )
+
+
+# ----- #4 재생성 사유 집계: 단독낙찰 이전 라운드를 미참여로 오집계하지 않음 -----
+async def test_awarded_prior_round_not_counted_as_no_show(clean):
+    """체인에 '단독낙찰'(preferred_sp_yn=True) 이전 라운드가 있어도, 이후 라운드의 미참여 재생성 예산을 소진하지 않는다.
+    (구버전: equal_bid_yn=False 인 단독낙찰 라운드를 미참여로 세어 round2 재생성이 막혔다.)"""
+    engine = clean
+    number = "C-AWARDED-PRIOR"
+    # round 1: 단독낙찰로 마감(preferred_sp_yn=True). 수동 재생성 등으로 체인이 이어진 상황을 가정.
+    await _seed_quotation(
+        engine, number=number, round_=1, status=QuotationStatus.CLOSED.value,
+        preferred_sp_yn=True, equal_bid_yn=False,
+    )
+    # round 2: 전원 미참여 → 미참여 재생성이 일어나야 한다(round 1 은 미참여로 세면 안 됨)
+    qt2 = await _seed_quotation(engine, number=number, round_=2, status=QuotationStatus.ACTIVE.value)
+    await _add_session(engine, qt2, status=SessionStatus.CREATED.value)
+
+    service = QuotationService(QuotationCRUD())
+    outcome = await service.close_and_decide(qt2)
+
+    rounds = await _rounds(engine, number)
+    round_numbers = [r.round for r in rounds]
+    assert outcome == CloseOutcome.REGENERATED, (
+        f"단독낙찰 이전 라운드는 미참여 예산을 소진하지 않아 round2 가 재생성돼야 함, 실제 {outcome}"
+    )
+    assert round_numbers == [1, 2, 3], f"round 3 이 생성돼야 함, 실제 {round_numbers}"
+
+
+# ----- #4 대비: 실제 미참여 이전 라운드는 예산을 소진(한도 1) -----
+async def test_no_show_prior_round_consumes_budget(clean):
+    """이전 라운드가 '미참여 재생성'(preferred_sp_yn=False, equal_bid_yn=False)이면 예산(1)을 소진 →
+    다음 라운드의 미참여는 재생성 없이 그냥 마감된다."""
+    engine = clean
+    number = "C-NOSHOW-PRIOR"
+    # round 1: 미참여로 마감(양성 표식) → no_part 예산 1 소진
+    await _seed_quotation(
+        engine, number=number, round_=1, status=QuotationStatus.CLOSED.value,
+        preferred_sp_yn=False, equal_bid_yn=False,
+    )
+    # round 2: 또 전원 미참여 → 한도 도달이라 재생성 없이 그냥 마감
+    qt2 = await _seed_quotation(engine, number=number, round_=2, status=QuotationStatus.ACTIVE.value)
+    await _add_session(engine, qt2, status=SessionStatus.CREATED.value)
+
+    service = QuotationService(QuotationCRUD())
+    outcome = await service.close_and_decide(qt2)
+
+    rounds = await _rounds(engine, number)
+    assert outcome == CloseOutcome.CLOSED, f"미참여 예산 소진 → 그냥 마감이어야 함, 실제 {outcome}"
+    assert [r.round for r in rounds] == [1, 2], "재생성되면 안 됨(round 3 없음)"

negodata/backend/tests/test_features.py

@@ -4,6 +4,8 @@ create(재조회로 created_at 적재) + list + get 경로를 라이브 DB 로
 """
 import uuid
 
+from common.enums import QuotationStatus, QuotationType
+
 
 async def _headers(client, company_id, login_id):
     await client.post(
@@ -47,22 +49,27 @@ async def test_quotation_setting_crud(client, company_id):
 
 async def test_quotation_create(client, company_id):
     h = await _headers(client, company_id, "qtuser")
+    # type/status 는 int 코드(QuotationType/QuotationStatus). number 는 서버가 생성하므로 미전송.
     body = {
         "qt_setting_id": str(uuid.uuid4()),
         "version_id": str(uuid.uuid4()),
         "name": "견적A",
-        "number": "Q-001",
+        "type": QuotationType.REQUOTE.value,
-        "type": "재견적",
+        "status": QuotationStatus.ACTIVE.value,
-        "status": "진행중",
         "start_time": "2026-06-16T00:00:00",
         "end_time": "2026-06-17T00:00:00",
     }
     r = await client.post("/v1/quotation/create", json=body, headers=h)
     res = r.json()
+    # 생성 응답은 본문(quotation)을 안 주고 qt_id/session_count 만 반환 → qt_id 로 재조회한다.
     assert res["result"]["success"] is True
-    q = res["quotation"]
+    qt_id = res["qt_id"]
+    assert qt_id
+
+    r = await client.get(f"/v1/quotation/{qt_id}", headers=h)
+    q = r.json()["quotation"]
     assert q["name"] == "견적A"
-    assert q["created_at"]  # 재조회 픽스
+    assert q["created_at"]  # 재조회로 created_at 적재 확인
 
     r = await client.get("/v1/quotation/list", headers=h)
     assert r.json()["total"] >= 1

negodata/backend/tests/test_item.py

@@ -7,6 +7,8 @@ import uuid
 import pytest_asyncio
 from sqlalchemy import text
 
+from common.enums import CompanyStatus
+
 
 async def _headers(client, company_id, login_id="itemuser", pw="pw1234"):
     await client.post(
@@ -21,9 +23,10 @@ async def _headers(client, company_id, login_id="itemuser", pw="pw1234"):
 async def other_company_id(db_engine) -> str:
     cid = uuid.uuid4()
     async with db_engine.begin() as conn:
+        # status 는 NOT NULL(모델 default 는 ORM 전용이라 raw INSERT 엔 안 먹음) → 명시.
         await conn.execute(
-            text("INSERT INTO companies (company_id, name) VALUES (:cid, :name)"),
+            text("INSERT INTO companies (company_id, name, status) VALUES (:cid, :name, :status)"),
-            {"cid": cid, "name": "다른회사"},
+            {"cid": cid, "name": "다른회사", "status": CompanyStatus.ACTIVE.value},
         )
     return str(cid)
 

negodata/front/src/lib/useScrollLock.ts

@@ -6,6 +6,7 @@ import { useEffect } from 'react';
 // 중첩 오버레이 대비 카운터로 관리 — 마지막 하나가 닫힐 때만 원복.
 let lockCount = 0;
 let prevOverflow = '';
+let pendingRestore = 0; // requestAnimationFrame id (0 = 없음)
 
 function scroller(): HTMLElement {
   return (document.scrollingElement as HTMLElement | null) ?? document.documentElement;
@@ -14,7 +15,13 @@ function scroller(): HTMLElement {
 export function useScrollLock(active = true) {
   useEffect(() => {
     if (!active) return;
-    if (lockCount === 0) {
+    if (pendingRestore !== 0) {
+      // 직전 해제로 예약된 복원이 남아 있으면 취소 — 언마운트→재마운트(라운드 전환 등) 사이의
+      // 일시적 0 을 흡수한다. 이때 overflow 는 아직 'hidden' 이고 prevOverflow 도 원래 값 그대로다.
+      cancelAnimationFrame(pendingRestore);
+      pendingRestore = 0;
+    } else if (lockCount === 0) {
+      // 진짜 첫 잠금일 때만 원래 overflow 를 보관하고 잠근다('hidden' 을 prevOverflow 로 캡처하는 사고 방지).
       const el = scroller();
       prevOverflow = el.style.overflow;
       el.style.overflow = 'hidden';
@@ -23,7 +30,11 @@ export function useScrollLock(active = true) {
     return () => {
       lockCount -= 1;
       if (lockCount === 0) {
-        scroller().style.overflow = prevOverflow;
+        // 마지막 해제는 다음 프레임으로 미룬다 — 곧바로 새 잠금이 들어오면(재마운트) 위에서 취소된다.
+        pendingRestore = requestAnimationFrame(() => {
+          pendingRestore = 0;
+          if (lockCount === 0) scroller().style.overflow = prevOverflow;
+        });
       }
     };
   }, [active]);

negodata/front/src/pages/quotation.tsx

@@ -1,4 +1,5 @@
 import { Settings, Plus } from 'lucide-react';
+import { keepPreviousData } from '@tanstack/react-query';
 import { useOverlayRouter } from '@/lib/useOverlayRouter';
 import { PageContainer } from '@/components/layout/PageContainer';
 import { PageToolbar, SearchInput } from '@/components/layout/PageToolbar';
@@ -51,8 +52,11 @@ export default function QuotationPage() {
 
   // 상세 요약은 리스트에서 find 하지 않고 단건 API 로 받아온다(딥링크 시 리스트 의존 제거).
   // 탭 복귀 시 재조회(자리비운 사이 스케줄러가 마감/낙찰/재생성했을 수 있음). 전역 기본은 false라 상세만 켠다.
+  // 라운드 전환(detailId 교체) 시 새 데이터 도착 전까지 이전 견적을 유지한다.
+  // 이렇게 해야 activeQuotation 이 잠시 null 로 떨어지지 않아 시트(key={qt_id})가 언마운트→재마운트되지 않고,
+  // 그 사이 useScrollLock 이 풀려 배경이 스크롤되는 현상도 사라진다. (목록 등 다른 쿼리와 동일한 패턴)
   const detailQuery = useGetQuotation(detailId ?? '', {
-    query: { enabled: !!detailId, refetchOnWindowFocus: true },
+    query: { enabled: !!detailId, refetchOnWindowFocus: true, placeholderData: keepPreviousData },
   });
   const activeQuotation = detailQuery.data?.quotation ?? null;