auto-stock/backend/app/services/db_worker.py

"""Background DB Writer Worker

ARCHITECTURE (CLAUDE.md 3.2):
  - 실시간/콜백 스레드에서는 절대 DB 세션을 직접 열지 않는다.
  - 모든 DB 쓰기 작업은 이 모듈의 db_task_queue 로 위임한다.
  - 워커가 큐를 소비하며 단일 세션에서 순차 처리 → 락 경합 제거.

Supported task types:
  - insert_trade_history       : dict  → TradeHistory INSERT
  - update_trade_history       : dict  → TradeHistory UPDATE by id (체결 확인 후 슬리피지/수량 반영)
  - update_trade_history_by_order_id : dict → TradeHistory UPDATE by order_id (KIS 주문번호 기준)
  - upsert_stock_analysis      : dict  → StockAnalysis single UPSERT
  - bulk_upsert_stock_analysis : list  → StockAnalysis batch UPSERT
  - upsert_stock_ban           : dict  → StockBan UPSERT (loss_tracker)
  - insert_news                 : dict  → News INSERT (뉴스 감성 분석 결과 저장)
  - log_mcp_deny               : dict  → MCP 거부 이벤트 로그 (DB 저장 없이 로그만)
  - upsert_portfolio           : dict  → Portfolio INSERT/UPDATE (strategy_type 포함)
  - upsert_active_trade        : dict  → ActiveTrade UPSERT (strategy_name 포함)
  - close_active_trade         : dict  → ActiveTrade 상태를 'closed'로 변경
  - upsert_stock_ban_v2        : dict  → StockBanV2 UPSERT (user_id, stock_code 스코프 — 멀티테넌트)

멀티테넌트(사용자 스코프) 확장 (docs/multi_tenant_design.md §3.4):
  - insert_trade_history / update_trade_history(_by_order_id) / upsert_portfolio /
    upsert_active_trade / close_active_trade / insert_trade_journal 는 payload
    data["user_id"] 가 있으면 조회 필터·생성 컬럼에 반영하고, 없으면(None) 기존과
    100% 동일하게 동작한다(하위 호환). 새 task 타입을 추가하지 않는다.
  - stock_bans 는 PK가 stock_code 단일이라 사용자 스코프로 확장 불가 → 신규 테이블
    stock_bans_v2(user_id, stock_code UNIQUE 쌍) + 신규 task upsert_stock_ban_v2 사용.
"""
import asyncio
import json
from app.core.json_helpers import _dumps
import logging
import time
import uuid
from datetime import date, datetime, timedelta
from pathlib import Path

from sqlalchemy import or_, select

from app.core.database import AsyncSessionLocal
from app.models.models import TradeHistory
from app.core.timeutil import now_kst

logger = logging.getLogger(__name__)

# ─────────────────────────────────────────────────────────────────────────────
# 전략명 상수 / 필터 헬퍼 (CLAUDE.md §8.5 항목 3·6)
#
# DT_STRATEGY_NAMES 는 mdd_manager.DT_STRATEGY_NAMES 와 **동일 목록**이어야 한다.
# 여기(db_worker)를 정본으로 두는 이유:
#   - db_worker 는 어떤 서비스 모듈도 모듈 수준에서 import 하지 않아 순환 참조가 없다.
#   - ActiveTrade 를 실제로 쓰는 계층이므로 필터 표현을 한 곳에 모을 수 있다.
# mdd_manager 쪽은 담당 에이전트가 달라 이번에 수정하지 않는다(후속 과제).
# ─────────────────────────────────────────────────────────────────────────────
DT_STRATEGY_NAMES: tuple[str, ...] = ("day_trade", "daytrading")


def not_dt_strategy_clause(column):
    """DT 계열 전략을 제외하는 SQL 3값 논리 안전 필터.

    `column != "day_trade"` 는 column IS NULL 인 legacy 행까지 함께 탈락시킨다
    (SQL 3값 논리: NULL != 'x' → UNKNOWN → WHERE 에서 거짓 취급).
    → NULL 행을 명시적으로 포함시킨 뒤 DT 계열만 NOT IN 으로 제외한다.

    Args:
        column: ActiveTrade.strategy_name 등 비교 대상 컬럼

    Returns:
        SQLAlchemy boolean expression
    """
    return or_(column.is_(None), column.notin_(DT_STRATEGY_NAMES))


# upsert_active_trade payload 중 ORM 컬럼이 아닌 '메타' 키 (setattr/생성 인자 제외 대상)
_ACTIVE_TRADE_META_KEYS = (
    "remaining_pct_multiply",
    "strategy_names",
    "strategy_name_not",
    # [M-A 2026-10-02] 멱등화 마커 — 워커가 처리 중에 채운다(ORM 컬럼 아님, 호출부는 넣지 않는다).
    "_target_id",
    "_applied_remaining_pct",
    # [2026-10-08] 스윙 진입가 정합 — 추가매수 가중평균 / 체결가 보정 (ORM 컬럼 아님)
    "accumulate_entry",
    "entry_fill_adjust",
    "_applied_entry_price",
    "_applied_quantity",
    "_retried_exec",  # [2026-10-09] 재시도 실행 표식(경고용, ORM 컬럼 아님)
)
# 정상 경로 재시도·DLQ 저장 전에 지우는 마커. `_applied_remaining_pct` 는 "이번 실행이
# 계산한 절대값"이라, 실패 후 재시도(큐 뒤로 재등록 → 그 사이 다른 multiply 가 먼저 적용될 수
# 있음)에서 재사용하면 그 갱신을 덮어쓴다. `_target_id`(close 대상 고정)는 유지한다 —
# 재시도도 처음 고른 행을 닫는 것이 맞다.
_RETRY_STRIP_MARKERS: tuple[str, ...] = (
    "_applied_remaining_pct", "_applied_entry_price", "_applied_quantity",
)
_PINNED_TASK_TYPES: tuple[str, ...] = ("close_active_trade", "upsert_active_trade")


def _strip_retry_markers(task: dict) -> None:
    data = task.get("data")
    if isinstance(data, dict):
        for k in _RETRY_STRIP_MARKERS:
            data.pop(k, None)


def _apply_strategy_name_filter(query, data: dict, column):
    """payload 의 전략명 필터 키를 SELECT 쿼리에 적용.

    지원 키 (우선순위 순 — 상위 키가 있으면 하위 키는 무시):
      1) strategy_name      : str  → column == value        (기존 동작, 무변경)
      2) strategy_names     : list → column IN (...)
      3) strategy_name_not  : str | list → column IS NULL OR column NOT IN (...)

    세 키가 모두 없거나 비어 있으면 query 를 **그대로** 반환하므로,
    새 키를 쓰지 않는 기존 호출부의 동작은 수정 전과 완전히 동일하다.
    """
    strategy_name = data.get("strategy_name")
    if strategy_name:
        return query.where(column == strategy_name)

    names = data.get("strategy_names")
    if names:
        _names = [n for n in names if n]
        if _names:
            return query.where(column.in_(_names))
        return query

    excluded = data.get("strategy_name_not")
    if excluded:
        if isinstance(excluded, str):
            excluded = [excluded]
        _excluded = [n for n in excluded if n]
        if _excluded:
            return query.where(or_(column.is_(None), column.notin_(_excluded)))
    return query


# ─────────────────────────────────────────────────────────────────────────────
# DLQ (Dead Letter Queue) — 3회 재시도 실패 시 파일로 백업
# ─────────────────────────────────────────────────────────────────────────────
DLQ_DIR = Path.home() / ".cache" / "auto-stock" / "dlq"


# 종료 drain 시간 예산(초). 초과분은 커밋을 포기하고 DLQ 로 보관한다.
DRAIN_DEADLINE_SECONDS = 20
# DLQ 재처리 완료 파일 이동 위치 — 재처리 무한루프 방지
DLQ_PROCESSED_DIR = DLQ_DIR / "processed"
# [C2 2026-09-30] 나이·충돌로 재큐잉하지 않은 파일 격리 위치 (증거 보존 — 삭제하지 않는다)
DLQ_EXPIRED_DIR = DLQ_DIR / "expired"
# [M-B 2026-10-02] 자동 재큐잉하면 안 되고 사람이 판단해야 하는 파일 (커밋 여부 불명 + 검증 불가)
DLQ_MANUAL_DIR = DLQ_DIR / "manual"

# [C2] 나이 제한 — replay_dlq() 가 failed_at 을 읽지 않아 **재기동마다 오래된 상태를
# 되살렸다.** 예: 9/25 drain 으로 빠진 upsert_active_trade 가 그 사이 전량 청산돼
# status='closed' 가 된 뒤 9/30 기동에 재큐잉되면, 핸들러가 active 행을 못 찾고
# 필수 키가 다 있으니 **새 active 행을 INSERT** 한다 → §8.4.1 의 "9/23 유령 행 129개"와
# 같은 실패 모드가 자동화된다. upsert_portfolio 는 이미 매도된 종목의 Portfolio 행을
# 재생성해 스윙 청산이 없는 수량을 팔려 한다(주문거부·과다매도).
#
# 타입별 차등: 위험의 성질이 다르기 때문이다.
#  - **상태 덮어쓰기**(active_trade/portfolio): payload 가 '그때의 현재 상태'다. 시간이
#    지나면 진실이 아니라 거짓이 되므로 짧게 잡는다. DT 포지션은 당일 청산(15:20 EOD)이
#    원칙이라 두 시간이면 이미 다른 국면이다.
#  - **기록 추가**(trade_history/journal/analysis/ban/news): 늦게 들어가도 '없던 기록이
#    생기는' 것이라 상태를 거짓으로 만들지 않는다. 다만 무한정 허용하면 과거 시각 행이
#    집계(get_uncovered_trades(days=7)·누적 실현손익)를 오염시키므로 24시간으로 끊는다.
# 상태 덮어쓰기 계열 — payload 가 '그때의 현재 상태'라 오래되면 거짓이 된다.
# [M-4 2026-09-30] close_active_trade 를 **여기서 뺐다.** 성격이 다르다: "닫아라"는
# 지시는 시간이 지나도 거짓이 되지 않는다(이미 닫혀 있으면 no-op, 아직 열려 있으면
# 닫는 것이 맞다). 오히려 격리하는 쪽이 **반대 방향의 유령 행**을 만든다 — 야간
# 재기동으로 close 가 격리되면 이미 청산된 포지션의 ActiveTrade 가 active 로 영구
# 잔존하고(§8.4.1 '유령 행 129개'와 같은 형태), 그 잔량이 다른 안전장치(유입 잔고
# 격리의 '시스템 보유' 하한)로 흘러든다. portfolio.py 의 정합 루프도 active 행이
# 2개 이상이면 continue 하므로 자동 복구되지 않는다.
_DLQ_STATE_TASK_TYPES: tuple[str, ...] = (
    "upsert_active_trade",
    "upsert_portfolio",
)
# 현재 상태와의 충돌을 확인할 타입 (나이 분류와 별개 — close 는 충돌 확인만 받는다)
_DLQ_CONFLICT_CHECK_TYPES: tuple[str, ...] = (
    "upsert_active_trade",
    "upsert_portfolio",
    "close_active_trade",
    # [M-B 2026-10-02] 같은 order_id 행이 이미 있으면 중복 INSERT
    "insert_trade_history",
)
# 상태 덮어쓰기 계열: 같은 날(로컬 날짜) + 2시간 이내만 재큐잉
DLQ_REPLAY_MAX_AGE_STATE_SECONDS: float = 2 * 3600
# 기록 추가 계열: 24시간 이내만 재큐잉
DLQ_REPLAY_MAX_AGE_RECORD_SECONDS: float = 24 * 3600
# [M-4] close_active_trade: 대폭 완화(7일). 무제한으로 두지 않는 이유는 충돌 판정이
# 불가한 상황(DB 미가동 → M-2 단락)에서 '새 포지션을 닫을' 위험이 남기 때문이다.
# 충돌 판정이 불가하면 아래에서 기록 계열 상한(24시간)으로 되돌린다.
DLQ_REPLAY_MAX_AGE_CLOSE_SECONDS: float = 7 * 24 * 3600

# [M-2 2026-09-30] 충돌 판정 단락 플래그. replay_dlq() 는 lifespan 에서 await 되므로
# 기동을 막는다. DB 가 아직 안 뜬 상태에서 파일마다 접속을 재시도하면 기동 지연이
# (파일 수 × 접속 타임아웃)까지 늘어난다 → 첫 실패 후에는 충돌 판정을 건너뛰고
# **나이 제한만** 적용한다(나이 제한은 단락 후에도 계속 적용된다).
_dlq_conflict_check_available: bool = True
# 한 번의 기동에서 재큐잉할 최대 DLQ 파일 수 (큐 폭주 방지)
DLQ_REPLAY_MAX_FILES = 500


# [M5 2026-09-28] DLQ 재처리 시 타입 복원.
# _save_to_dlq 는 _dumps(..., default=str) (내부적으로 json.dumps) 를 쓰므로 Decimal → "1234.5",
# datetime/date → ISO 문자열로 **열화**된다. 그 dict 를 그대로 큐에 되돌리면
# `TradeHistory(**data)` 같은 ORM 대입에서 asyncpg 가 DateTime 컬럼의 str 을 거부해
# 3회 재시도 후 **새 DLQ 파일로 다시 쌓인다**(기동마다 증식). 재큐잉 전에 모델 컬럼
# 타입을 보고 되돌린다.
_DLQ_TASK_MODELS: dict[str, str] = {
    "insert_trade_history": "TradeHistory",
    "update_trade_history": "TradeHistory",
    "update_trade_history_by_order_id": "TradeHistory",
    "upsert_portfolio": "Portfolio",
    "upsert_active_trade": "ActiveTrade",
    "close_active_trade": "ActiveTrade",
    "upsert_stock_analysis": "StockAnalysis",
    "bulk_upsert_stock_analysis": "StockAnalysis",
    "insert_trade_journal": "TradeJournal",
    "upsert_stock_ban": "StockBan",
    "upsert_stock_ban_v2": "StockBanV2",
    "insert_news": "News",
}


def _coerce_value_for_column(value, col_type):
    """컬럼 타입에 맞춰 문자열을 원래 타입으로 되돌린다. 실패 시 원본 그대로."""
    if not isinstance(value, str) or not value:
        return value
    try:
        from sqlalchemy import DateTime, Date, Numeric, Float, Integer, Boolean
        if isinstance(col_type, DateTime):
            return datetime.fromisoformat(value)
        if isinstance(col_type, Date):
            return date.fromisoformat(value)
        if isinstance(col_type, Numeric) and not isinstance(col_type, Float):
            from decimal import Decimal
            return Decimal(value)
        if isinstance(col_type, Float):
            return float(value)
        if isinstance(col_type, Integer):
            return int(value)
        if isinstance(col_type, Boolean):
            return value.lower() in ("true", "1", "yes")
    except (ValueError, TypeError, ArithmeticError):
        # 되돌리지 못하면 원본을 유지 — 핸들러가 판단하게 한다(조용히 버리지 않는다)
        logger.warning("[DLQ] 타입 복원 실패 — 원본 유지: %r (%s)", value, col_type)
    return value


def _restore_dlq_task_types(task: dict) -> dict:
    """[M5] DLQ 파일에서 읽은 task 의 payload 타입을 모델 스키마 기준으로 복원."""
    model_name = _DLQ_TASK_MODELS.get(task.get("type", ""))
    if not model_name:
        return task
    try:
        from app.models import models as _m
        model = getattr(_m, model_name, None)
        if model is None:
            return task
        cols = {c.key: c.type for c in model.__table__.columns}
    except Exception as e:
        logger.warning("[DLQ] 모델 스키마 조회 실패 (%s): %s", model_name, e)
        return task

    def _fix(d):
        if not isinstance(d, dict):
            return d
        return {
            k: (_coerce_value_for_column(v, cols[k]) if k in cols else v)
            for k, v in d.items()
        }

    data = task.get("data")
    if isinstance(data, list):
        task["data"] = [_fix(row) for row in data]
    elif isinstance(data, dict):
        task["data"] = _fix(data)
    # update 계열은 갱신 필드를 별도 키에 담는다
    if isinstance(task.get("updates"), dict):
        task["updates"] = _fix(task["updates"])
    return task


def _dlq_max_age_seconds(task_type: str) -> float:
    """타입별 재큐잉 허용 나이(초).

    close_active_trade 는 충돌 판정이 **가능할 때만** 7일까지 허용한다. 판정이 단락된
    상태(DB 미가동)에서는 '새 포지션을 닫을' 위험이 남으므로 기록 계열 상한으로 내린다.
    """
    if task_type in _DLQ_STATE_TASK_TYPES:
        return DLQ_REPLAY_MAX_AGE_STATE_SECONDS
    if task_type == "close_active_trade":
        if _dlq_conflict_check_available:
            return DLQ_REPLAY_MAX_AGE_CLOSE_SECONDS
        return DLQ_REPLAY_MAX_AGE_RECORD_SECONDS
    return DLQ_REPLAY_MAX_AGE_RECORD_SECONDS


def _dlq_age_verdict(entry: dict, task_type: str) -> tuple[bool, str]:
    """[C2] failed_at 기준 나이 판정. (재큐잉 허용?, 사유) 반환.

    ``failed_at`` 은 ``_save_to_dlq`` 가 ``datetime.now().isoformat()`` (naive 로컬)로
    쓴다. 같은 시계로 비교해야 하므로 여기서도 ``datetime.now()`` 를 쓴다
    (``now_kst()`` 와 섞으면 컨테이너 TZ 에 따라 9시간 오차가 난다).

    나이를 알 수 없으면: 상태 덮어쓰기 계열은 **격리**(보수적), 기록 계열은 허용.
    """
    raw = entry.get("failed_at")
    failed_at = None
    if isinstance(raw, str) and raw:
        try:
            failed_at = datetime.fromisoformat(raw)
        except ValueError:
            failed_at = None

    if failed_at is None:
        if task_type in _DLQ_STATE_TASK_TYPES:
            return False, f"failed_at 없음/파싱불가({raw!r}) — 상태 덮어쓰기 계열은 보수적 격리"
        # close_active_trade 는 나이를 몰라도 허용한다(닫으라는 지시는 늦어도 유효).
        # 새 포지션을 닫는 위험은 충돌 판정(created_at > failed_at)이 막는다.
        return True, "failed_at 없음 — 기록/close 계열이라 허용"

    if failed_at.tzinfo is not None:
        failed_at = failed_at.replace(tzinfo=None)
    now = datetime.now()
    age = (now - failed_at).total_seconds()
    limit = _dlq_max_age_seconds(task_type)
    if age > limit:
        return False, (
            f"나이 {age / 3600:.1f}시간 > 상한 {limit / 3600:.1f}시간 "
            f"(failed_at={failed_at.isoformat()})"
        )
    if task_type in _DLQ_STATE_TASK_TYPES and failed_at.date() != now.date():
        # 2시간 이내라도 날짜가 다르면 다른 세션이다(자정 전후 기동)
        return False, (
            f"상태 덮어쓰기 계열인데 날짜가 다름 "
            f"(failed_at={failed_at.date()}, today={now.date()})"
        )
    return True, f"나이 {age / 60:.0f}분"


async def _dlq_state_conflict(task: dict, entry: dict) -> tuple[bool, str]:
    """[C2] 상태 덮어쓰기 계열이 **현재 상태와 충돌**하는지 읽기 전용으로 확인.

    나이 제한을 통과해도(≤2시간) 그 사이 청산·재진입이 있었으면 유령 행을 만든다.
    여기서는 SELECT 만 수행하고 ``commit()`` 을 호출하지 않는다 — 기동 시 1회,
    실시간 계층이 아니다(CLAUDE.md §3.2 허용예외③ 와 동일한 성격). **쓰기는 여전히
    큐 경유**다.

    판정 실패(DB 미가동 등)는 '충돌 아님'으로 돌려 나이 제한만 적용한다 — 여기서
    예외로 죽으면 기동이 막힌다.

    Returns:
        tuple[bool, str]: (충돌인가?, 사유)
    """
    global _dlq_conflict_check_available

    task_type = task.get("type", "")
    data = task.get("data") or {}
    stock_code = data.get("stock_code") if isinstance(data, dict) else None
    if not stock_code or task_type not in _DLQ_CONFLICT_CHECK_TYPES:
        return False, ""
    # [M-2] 이미 판정 불가로 확인됐으면 더 시도하지 않는다(기동 지연 유계화).
    # 로그도 남기지 않는다 — 첫 실패에서 한 번만 남겼다.
    if not _dlq_conflict_check_available:
        return False, ""

    try:
        from app.models.models import ActiveTrade, Portfolio

        if task_type == "insert_trade_history":
            dup = await _dlq_trade_history_exists(data, entry)
            if dup:
                return True, f"{stock_code}: 같은 order_id 의 TradeHistory 가 이미 있음 → 중복 INSERT 위험"
            return False, ""

        async with AsyncSessionLocal() as session:
            active = (
                await session.execute(
                    select(ActiveTrade)
                    .where(
                        ActiveTrade.stock_code == stock_code,
                        ActiveTrade.status == "active",
                    )
                    .order_by(ActiveTrade.id.desc())
                    .limit(1)
                )
            ).scalars().first()
            pf = (
                await session.execute(
                    select(Portfolio).where(Portfolio.stock_code == stock_code).limit(1)
                )
            ).scalars().first()
            pf_qty = int(getattr(pf, "quantity", 0) or 0) if pf is not None else 0

            if task_type == "upsert_active_trade":
                # active 행도 없고 보유 수량도 0 → 재큐잉하면 유령 active 행이 생긴다
                if active is None and pf_qty <= 0:
                    return True, (
                        f"{stock_code}: active 행 없음 + 보유수량 0 → 유령 ActiveTrade 생성 위험"
                    )
            elif task_type == "upsert_portfolio":
                # Portfolio 행이 없으면 이미 정리된 종목이다. sync_portfolio() 가 KIS
                # 잔고에서 매 주기 재생성하므로 스킵해도 실보유는 복구된다.
                if pf is None:
                    return True, f"{stock_code}: Portfolio 행 없음 → 매도된 종목 재생성 위험"
            elif task_type == "close_active_trade":
                # failed_at 이후에 생긴 active 행이면 '새 포지션'이다 — 그걸 닫으면 안 된다
                raw = entry.get("failed_at")
                try:
                    _fa = datetime.fromisoformat(raw) if isinstance(raw, str) else None
                except ValueError:
                    _fa = None
                if _fa is not None and _fa.tzinfo is not None:
                    _fa = _fa.replace(tzinfo=None)
                _created = getattr(active, "created_at", None) if active is not None else None
                if _created is not None and _fa is not None:
                    if _created.tzinfo is not None:
                        _created = _created.replace(tzinfo=None)
                    if _created > _fa:
                        return True, (
                            f"{stock_code}: 현재 active 행이 failed_at 이후 생성"
                            f"(created_at={_created.isoformat()}) → 새 포지션을 닫을 위험"
                        )
    except Exception as e:
        # [M-2] 첫 실패에서 단락한다 — 이후 파일은 접속을 재시도하지 않는다.
        _dlq_conflict_check_available = False
        logger.warning(
            "[DLQ] 상태 충돌 확인 실패 (%s/%s): %s — 이후 파일은 충돌 판정을 건너뛰고 "
            "나이 제한만 적용한다(기동 지연 방지). close_active_trade 허용 나이도 "
            "%.0f시간으로 내려간다.",
            task_type, stock_code, e, DLQ_REPLAY_MAX_AGE_RECORD_SECONDS / 3600,
        )
        return False, ""
    return False, ""


def _parse_dt(value) -> datetime | None:
    if isinstance(value, datetime):
        return value.replace(tzinfo=None) if value.tzinfo else value
    if isinstance(value, str) and value:
        try:
            v = datetime.fromisoformat(value)
            return v.replace(tzinfo=None) if v.tzinfo else v
        except ValueError:
            return None
    return None


def _dlq_ref_day_bounds(data: dict, entry: dict) -> tuple[datetime, datetime]:
    """[L2] 주문번호 중복 확인 기준일의 [00:00, 다음날 00:00) — `_order_id_ref_date` 와 같은 우선순위.

    payload 에 날짜 단서(order_date/created_at/executed_at)가 없으면 DLQ failed_at 날짜를 쓴다
    (`_order_id_ref_date` 의 '오늘' 폴백은 다음 날 재처리 때 날짜를 틀리게 잡는다).
    """
    if any(data.get(k) not in (None, "") for k in ("order_date", "created_at", "executed_at")):
        day = _order_id_ref_date(data)
    else:
        day = (_parse_dt(entry.get("failed_at")) or now_kst()).date()
    start = datetime(day.year, day.month, day.day)
    return start, start + timedelta(days=1)


async def _dlq_trade_journal_exists(data: dict, entry: dict) -> bool | None:
    """[L3] 커밋 여부 불명 TradeJournal 이 이미 저장됐는가. True/False, 판정 불가면 None.

    - trade_id 가 있으면 그 trade_id 의 일지 존재 여부
    - order_id 만 있으면 같은 날 TradeHistory 로 trade_id 를 해소한 뒤 존재 여부.
      해소되지 않으면 None — 핸들러가 trade_id=None 일지를 새로 만들어 중복을 막을 수 없다.
    예외는 호출자에게 넘긴다(호출자가 manual 격리).
    """
    from app.models.models import TradeJournal

    trade_id = data.get("trade_id") or None
    async with AsyncSessionLocal() as session:
        if not trade_id and data.get("order_id"):
            start, end = _dlq_ref_day_bounds(data, entry)
            trade_id = (
                await session.execute(
                    select(TradeHistory.id)
                    .where(
                        TradeHistory.order_id == str(data["order_id"]),
                        TradeHistory.created_at >= start,
                        TradeHistory.created_at < end,
                    )
                    .order_by(TradeHistory.id.desc())
                    .limit(1)
                )
            ).scalar_one_or_none()
        if not trade_id:
            return None
        row = (
            await session.execute(
                select(TradeJournal.id).where(TradeJournal.trade_id == trade_id).limit(1)
            )
        ).first()
    return row is not None


async def _dlq_trade_history_exists(data: dict, entry: dict) -> bool:
    """[M-B] order_id 가 같은 TradeHistory 행이 이미 있는가 (SELECT 전용, commit 없음).

    KIS 주문번호는 **날마다 재사용**되므로 종목·주문유형·기준일(±1일) 로 범위를 좁힌다.
    기준일은 payload created_at(M-7 이후 항상 있음) → 없으면 DLQ failed_at.
    order_id 가 없으면 False (판정 불가 — 호출자가 사유에 따라 처리).
    예외는 호출자(_dlq_state_conflict)의 단락 처리로 넘긴다.
    """
    order_id = data.get("order_id") if isinstance(data, dict) else None
    if not order_id:
        return False
    start, end = _dlq_ref_day_bounds(data, entry)
    query = select(TradeHistory.id).where(
        TradeHistory.order_id == str(order_id),
        TradeHistory.created_at >= start,
        TradeHistory.created_at < end,
    )
    if data.get("stock_code"):
        query = query.where(TradeHistory.stock_code == data["stock_code"])
    if data.get("order_type"):
        query = query.where(TradeHistory.order_type == data["order_type"])
    async with AsyncSessionLocal() as session:
        row = (await session.execute(query.limit(1))).first()
    return row is not None


def _dlq_commit_unknown_requeue_basis(task: dict) -> str | None:
    """[M-B] 커밋 여부 불명 INSERT 를 재큐잉해도 되는 근거. 없으면 None (→ manual/ 격리).

    - insert_trade_history: order_id 가 있어 _dlq_state_conflict 가 중복을 확인할 수 있을 때
    - insert_trade_journal: trade_id 또는 order_id — 재큐잉 전에 `_dlq_trade_journal_exists` 로
      실제 존재를 확인한다(L3). 해소 불가면 manual.
    - insert_news: url 이 있으면 핸들러가 URL 중복을 건너뛴다
    """
    task_type = task.get("type")
    data = task.get("data") if isinstance(task.get("data"), dict) else {}
    if task_type == "insert_trade_history" and data.get("order_id"):
        return "order_id 중복 확인"
    if task_type == "insert_trade_journal" and (data.get("order_id") or data.get("trade_id")):
        return "trade_id/order_id 기준 일지 존재 확인"
    if task_type == "insert_news" and data.get("url"):
        return "핸들러의 URL 중복 스킵"
    return None


# [M-3] 직전 replay_dlq() 가 재큐잉한 태스크 타입 — 기동 시 밴 재로드 판단용
last_replay_requeued_types: set[str] = set()


def _count_json(d: Path) -> int:
    try:
        return sum(1 for f in d.glob("*.json") if f.is_file()) if d.exists() else 0
    except OSError:
        return -1


async def dlq_status_counts() -> dict[str, int]:
    """[L1] DLQ 파일 건수 (대기/수동확인/격리). 내용은 노출하지 않는다 — /health·알림용."""
    def _counts() -> dict[str, int]:
        return {
            "pending": _count_json(DLQ_DIR),
            "manual": _count_json(DLQ_MANUAL_DIR),
            "expired": _count_json(DLQ_EXPIRED_DIR),
        }

    return await asyncio.to_thread(_counts)


async def replay_dlq() -> int:
    """기동 시 DLQ 파일을 읽어 db_task_queue 에 재큐잉. 재큐잉한 건수 반환.

    [F4 2026-09-28] DLQ 는 지금까지 **쓰기 전용**이었다(읽는 코드 0건). 3회 재시도
    실패분과 종료 drain 미커밋분이 파일로만 남아 유실이 영구화되고, 복구 수단은
    수동 psql 뿐이었다.

    CLAUDE.md §3.2 에 따라 **직접 DB 쓰기를 하지 않고 반드시 큐를 경유**한다.
    파일은 재큐잉 **전에** ``processed/`` 로 옮긴다 — 옮긴 뒤 큐 투입이 실패해도
    파일은 보존되고(수동 복구 가능), 다음 기동에서 같은 태스크를 두 번 큐에 넣지
    않는다(insert_trade_history 는 멱등이 아니므로 중복이 더 위험하다).

    손상된 JSON 은 건너뛰고 warning 을 남긴다(파일은 그대로 두어 사람이 본다).

    [C2 2026-09-30] **나이 제한 + 상태 충돌 확인**을 거친다. 초기 판은 failed_at 을
    읽지 않아 재기동마다 오래된 상태를 되살렸다(유령 ActiveTrade/Portfolio 생성).
    재큐잉하지 않은 파일은 삭제하지 않고 ``expired/`` 로 격리하고 사유를 warning 으로
    남긴다. 정책 근거는 위 상수 주석 참조.
    """
    def _list_files() -> list:
        if not DLQ_DIR.exists():
            return []
        return sorted(f for f in DLQ_DIR.glob("*.json") if f.is_file())

    files = await asyncio.to_thread(_list_files)
    if not files:
        return 0

    if len(files) > DLQ_REPLAY_MAX_FILES:
        logger.error(
            "❌ [DLQ] 파일 %d건이 상한 %d건을 초과 — 앞의 %d건만 재처리하고 나머지는 남긴다",
            len(files), DLQ_REPLAY_MAX_FILES, DLQ_REPLAY_MAX_FILES,
        )
        files = files[:DLQ_REPLAY_MAX_FILES]

    logger.warning("[DLQ] 재처리 시작: %d건", len(files))
    # [M-2] 이번 기동의 충돌 판정 가용성 초기화 (이전 호출의 단락을 물려받지 않는다)
    global _dlq_conflict_check_available
    _dlq_conflict_check_available = True
    requeued = 0
    skipped = 0
    expired = 0
    manual = 0
    last_replay_requeued_types.clear()
    for f in files:
        try:
            raw = await asyncio.to_thread(f.read_text, encoding="utf-8")
            entry = json.loads(raw)
            task = entry.get("task") if isinstance(entry, dict) else None
            if not isinstance(task, dict) or not task.get("type"):
                logger.warning("[DLQ] 건너뜀(형식 불량): %s", f.name)
                skipped += 1
                continue

            task_type = task.get("type", "unknown")

            def _move_to(dest, path=f):
                dest.mkdir(parents=True, exist_ok=True)
                path.rename(dest / path.name)

            # [M-B 2026-10-02] 종료 시 처리 중이던 비멱등 INSERT(커밋 여부 불명)는
            # 중복 여부를 확인할 근거가 있을 때만 재큐잉한다. 근거가 없거나 DB 확인이
            # 불가하면 manual/ 로 격리한다(자동 재큐잉은 TradeHistory/Journal 중복을 만든다).
            if entry.get("error") == _DLQ_COMMIT_UNKNOWN_REASON and task_type == "log_mcp_deny":
                # [L4] 로그 전용 이벤트 — 재실행하면 중복 기록일 뿐 상태에 영향이 없다 → 재큐잉 안 함
                await asyncio.to_thread(_move_to, DLQ_EXPIRED_DIR)
                logger.warning("[DLQ] 격리(로그 전용, 커밋 여부 불명): %s", f.name)
                expired += 1
                continue
            if entry.get("error") == _DLQ_COMMIT_UNKNOWN_REASON and task_type == "insert_trade_journal":
                # [L3] 재큐잉 전 실제 존재 확인 — 이미 있으면 격리, 판정 불가(해소 실패·DB 장애)면 manual
                try:
                    _jexists = await _dlq_trade_journal_exists(task.get("data") or {}, entry)
                except Exception as _je:
                    logger.warning("[DLQ] 일지 존재 확인 실패: %s — %s", f.name, _je)
                    _jexists = None
                if _jexists is True:
                    await asyncio.to_thread(_move_to, DLQ_EXPIRED_DIR)
                    logger.warning("[DLQ] 격리(같은 trade_id 일지 이미 존재): %s", f.name)
                    expired += 1
                    continue
                if _jexists is None:
                    await asyncio.to_thread(_move_to, DLQ_MANUAL_DIR)
                    logger.error(
                        "❌ [DLQ] 수동 확인 필요(커밋 여부 불명 일지, 존재 확인 불가): %s → %s",
                        f.name, DLQ_MANUAL_DIR,
                    )
                    manual += 1
                    continue
            if entry.get("error") == _DLQ_COMMIT_UNKNOWN_REASON:
                _basis = _dlq_commit_unknown_requeue_basis(task)
                _needs_db = task_type == "insert_trade_history"
                if _basis is None or (_needs_db and not _dlq_conflict_check_available):
                    await asyncio.to_thread(_move_to, DLQ_MANUAL_DIR)
                    logger.error(
                        "❌ [DLQ] 수동 확인 필요(커밋 여부 불명, 중복 확인 근거 %s): %s | type=%s | "
                        "failed_at=%s → %s",
                        "없음" if _basis is None else "확인 불가(DB)", f.name, task_type,
                        entry.get("failed_at"), DLQ_MANUAL_DIR,
                    )
                    manual += 1
                    continue

            # [C2] ① 상태 충돌 확인 (읽기 전용) — 유령 행 생성 차단.
            # 나이 판정보다 **먼저** 한다: close_active_trade 의 허용 나이가 '충돌 판정
            # 가능 여부'에 달려 있어(M-4), 판정을 먼저 해야 첫 파일이 blind 로 통과하는
            # 창이 생기지 않는다. DB 장애 시에는 여기서 한 번만 시도하고 단락된다(M-2).
            _conflict, _creason = await _dlq_state_conflict(task, entry)
            if _conflict:
                await asyncio.to_thread(_move_to, DLQ_EXPIRED_DIR)
                logger.warning(
                    "[DLQ] 격리(현재 상태와 충돌): %s | type=%s | failed_at=%s | 사유=%s",
                    f.name, task_type, entry.get("failed_at"), _creason,
                )
                expired += 1
                continue
            # [M-B] 커밋 여부 불명 trade_history 인데 방금 중복 확인이 DB 장애로 단락됐다면
            # '충돌 아님'은 확인된 결과가 아니다 → 재큐잉 금지
            if (
                entry.get("error") == _DLQ_COMMIT_UNKNOWN_REASON
                and task_type == "insert_trade_history"
                and not _dlq_conflict_check_available
            ):
                await asyncio.to_thread(_move_to, DLQ_MANUAL_DIR)
                logger.error(
                    "❌ [DLQ] 수동 확인 필요(커밋 여부 불명 + 중복 확인 불가): %s | failed_at=%s → %s",
                    f.name, entry.get("failed_at"), DLQ_MANUAL_DIR,
                )
                manual += 1
                continue

            # [C2] ② 나이 제한 — 오래된 상태 재생이 유실보다 위험하다
            _ok, _why = _dlq_age_verdict(entry, task_type)
            if not _ok:
                await asyncio.to_thread(_move_to, DLQ_EXPIRED_DIR)
                logger.warning(
                    "[DLQ] 격리(재큐잉 안 함): %s | type=%s | failed_at=%s | 사유=%s",
                    f.name, task_type, entry.get("failed_at"), _why,
                )
                expired += 1
                continue

            await asyncio.to_thread(_move_to, DLQ_PROCESSED_DIR)
            # 재시도 카운터를 초기화해 재큐잉분도 3회 재시도를 받게 한다
            task.pop("_retry_count", None)
            # [M5] default=str 로 열화된 Decimal/datetime/date 를 되돌린다
            task = _restore_dlq_task_types(task)
            await enqueue_db_write(task)
            last_replay_requeued_types.add(task_type)
            logger.info(
                "[DLQ] 재큐잉: %s | type=%s | %s", f.name, task_type, _why,
            )
            requeued += 1
        except json.JSONDecodeError as e:
            logger.warning("[DLQ] 건너뜀(JSON 손상): %s — %s", f.name, e)
            skipped += 1
        except Exception as e:
            logger.error("❌ [DLQ] 재처리 실패: %s — %s", f.name, e, exc_info=True)
            skipped += 1

    logger.warning(
        "[DLQ] 재처리 완료: 재큐잉 %d건 / 격리 %d건 / 수동확인 %d건 / 건너뜀 %d건 "
        "(처리분→%s, 격리분→%s, 수동→%s)",
        requeued, expired, manual, skipped, DLQ_PROCESSED_DIR, DLQ_EXPIRED_DIR, DLQ_MANUAL_DIR,
    )
    return requeued


# [2026-10-09] 런타임 DLQ 적재/워커 적체 경보. 같은 키는 이 간격(초)에 1회만 텔레그램 발송.
DLQ_ALERT_INTERVAL_SEC = 300.0
WORKER_QUEUE_ALERT_DEPTH = 200          # qsize 가 이 값 이상이면 경보
DLQ_RUNTIME_REPLAY_INTERVAL_SEC = 300.0  # 런타임 pending DLQ 재처리 주기
CIRCUIT_OPEN_FAILURES = 5               # 연속 실패 이 횟수 이상이면 DB 장애로 간주(회로 열림)
CIRCUIT_RETRY_DELAY_SEC = 8.0
CIRCUIT_MAX_HOLD_SEC = 120.0            # 회로 열림 상태로 재시도 횟수를 아끼는 최대 시간
_alert_tasks: set = set()
# 워커 상태 스냅샷 (/health·경보용). 워커가 갱신한다.
_worker_state: dict = {"consecutive_failures": 0, "circuit_open": False, "last_success_ts": None}


async def _send_alert_impl(category: str, message: str) -> None:
    try:
        from app.services.telegram_notifier import telegram_notifier
        await asyncio.wait_for(telegram_notifier.notify_system_warning(category, message), timeout=10)
    except Exception as e:
        logger.warning("[DBWorker] 텔레그램 경보 전송 실패(%s): %s", category, e)


def _alert_throttled_v2(key: str, category: str, message: str,
                        interval: float = DLQ_ALERT_INTERVAL_SEC) -> bool:
    """throttle 된 비차단 텔레그램 경보. 이벤트 루프가 없으면 로그만 남긴다."""
    from app.core.log_throttle import should_log
    ok, suppressed = should_log(key, interval)
    if not ok:
        return False
    if suppressed:
        message = f"{message} (+{suppressed}건 억제)"
    try:
        fut = asyncio.ensure_future(_send_alert_impl(category, message))
    except RuntimeError as e:
        logger.warning("[DBWorker] 경보 태스크 생성 실패: %s", e)
        return False
    _alert_tasks.add(fut)
    fut.add_done_callback(_alert_tasks.discard)
    return True


QUEUE_DEPTH_CHECK_INTERVAL_SEC = 30.0   # 성공 경로 적체 점검 주기
SLOW_TASK_ALERT_SEC = 5.0               # 단일 쓰기가 이 시간 이상 걸리면 느린 DB 로 간주
_depth_check_state: dict = {"last": 0.0}


def _check_queue_backlog_v2(queue: asyncio.Queue, task_elapsed: float | None = None,
                            now: float | None = None) -> bool:
    """[2026-10-09] 성공 경로 적체 점검. 쓰기가 실패 없이 느린 경우에도 qsize 경보를 낸다.

    QUEUE_DEPTH_CHECK_INTERVAL_SEC 마다 1회만 평가(핫패스 비용 최소). 경보는 실패 경로와
    같은 키로 throttle 되어 중복 발송되지 않는다.
    """
    t = time.monotonic() if now is None else now
    if t - _depth_check_state["last"] < QUEUE_DEPTH_CHECK_INTERVAL_SEC:
        return False
    _depth_check_state["last"] = t
    depth = queue.qsize()
    if depth < WORKER_QUEUE_ALERT_DEPTH:
        return False
    slow = f", 직전 쓰기 {task_elapsed:.1f}s" if task_elapsed is not None else ""
    return _alert_throttled_v2(
        "db_worker:queue_depth", "DB_WORKER",
        f"DB 쓰기 큐 적체 qsize={depth} (실패 없음{slow}). DB 지연 확인 필요.",
    )


def get_worker_health_v2() -> dict:
    """큐 적체·회로 상태 (동기, DB 접근 없음). /health 노출용."""
    q = _db_task_queue
    return {
        "qsize": q.qsize() if q is not None else 0,
        "consecutive_failures": _worker_state["consecutive_failures"],
        "circuit_open": _worker_state["circuit_open"],
        "last_success_ts": _worker_state["last_success_ts"],
    }


async def _save_to_dlq(task: dict, error: str) -> None:
    """3회 재시도 실패한 task를 DLQ 파일로 저장.

    저장 위치: ~/.cache/auto-stock/dlq/{timestamp}_{task_type}_{uuid}.json
    파일 I/O는 asyncio.to_thread로 감싸 이벤트 루프 블로킹을 방지한다.
    """
    try:
        task_type = task.get("type", "unknown")
        filename = (
            f"{datetime.now().strftime('%Y%m%d_%H%M%S')}"
            f"_{task_type}"
            f"_{uuid.uuid4().hex[:8]}.json"
        )
        dlq_entry = {
            "task": task,
            "error": str(error),
            "failed_at": datetime.now().isoformat(),
            "retry_count": task.get("_retry_count", 0),
        }

        def _write():
            DLQ_DIR.mkdir(parents=True, exist_ok=True)
            (DLQ_DIR / filename).write_text(
                _dumps(dlq_entry, default=str),
                encoding="utf-8",
            )

        await asyncio.to_thread(_write)
        logger.critical(
            "[DLQ] Task saved to DLQ: %s | type=%s | error=%s",
            filename, task_type, error,
        )
        _alert_throttled_v2(
            "db_worker:dlq_saved", "DLQ",
            f"DB 쓰기 {task_type} 가 재시도 실패로 DLQ 에 적재됨 "
            f"(ActiveTrade·밴 등 DB 상태가 어긋날 수 있음). 주기 재처리가 복구를 시도한다.",
        )
    except Exception as e:
        logger.critical(
            "[DLQ] Failed to save to DLQ: %s | task_type=%s",
            e, task.get("type", "unknown"),
        )
        _alert_throttled_v2(
            "db_worker:dlq_save_failed", "DLQ",
            f"DLQ 파일 저장 자체가 실패 — 태스크 유실 (type={task.get('type', 'unknown')}): {e}",
        )


# ─────────────────────────────────────────────────────────────────────────────
# 전역 큐 — 앱 어디서나 import 하여 put() 호출
# [이슈8 수정] asyncio.Queue()를 모듈 로드 시점에 생성하면 이벤트 루프가 없는
# 환경(pytest, 스크립트 임포트 등)에서 RuntimeError가 발생할 수 있음.
# _db_task_queue를 None으로 두고, 첫 접근 시 lazy 초기화하는 패턴으로 변경.
# 외부 코드는 기존처럼 db_task_queue 이름으로 접근 가능 (get_db_task_queue() 내부 참조).
# ─────────────────────────────────────────────────────────────────────────────
_db_task_queue: asyncio.Queue | None = None
# [M-D] 워커가 종료 drain 을 끝낸 큐에 다는 표식 (큐 인스턴스 단위 — 테스트가 큐를 갈아끼우면 함께 사라진다)
_QUEUE_CLOSED_ATTR = "_autostock_worker_closed"


def get_db_task_queue() -> asyncio.Queue:
    """이벤트 루프 기동 후 최초 호출 시 Queue를 생성(lazy init)하여 반환."""
    global _db_task_queue
    if _db_task_queue is None:
        _db_task_queue = asyncio.Queue()
    return _db_task_queue


# 하위 호환: 기존 코드가 `db_task_queue.put()` 형태로 직접 참조하는 경우를 위해
# 프록시 객체 대신 get_db_task_queue()를 통해 접근하도록 모듈 수준 alias 제공.
# 단, 모듈 임포트 시점에 이벤트 루프가 없어도 오류가 발생하지 않음.
class _QueueProxy:
    """asyncio.Queue를 lazy init으로 감싸는 투명 프록시."""

    def __getattr__(self, name):
        return getattr(get_db_task_queue(), name)

    async def put(self, item):
        q = get_db_task_queue()
        # [M-D 2026-10-02] 워커가 종료 drain 을 마친 뒤의 투입은 아무도 소비하지 않는다
        # (shield 된 청산 마무리·백그라운드 태스크가 종료 중에 늦게 enqueue). 큐에 넣으면
        # 프로세스 종료와 함께 사라지므로 DLQ 파일로 직접 보관한다 → 다음 기동 replay_dlq.
        if getattr(q, _QUEUE_CLOSED_ATTR, False):
            logger.error(
                "❌ [DBWorker] 워커 종료 후 enqueue — DLQ 로 직접 보관 (type=%s)",
                item.get("type", "unknown") if isinstance(item, dict) else type(item).__name__,
            )
            await _save_to_dlq(item if isinstance(item, dict) else {"type": "unknown", "data": item},
                               "enqueued_after_worker_shutdown")
            return
        await q.put(item)

    async def get(self):
        return await get_db_task_queue().get()

    def task_done(self):
        get_db_task_queue().task_done()

    def qsize(self):
        return get_db_task_queue().qsize()

    def empty(self):
        return get_db_task_queue().empty()


db_task_queue: asyncio.Queue = _QueueProxy()  # type: ignore[assignment]


# ─────────────────────────────────────────────────────────────────────────────
# 편의 함수 — 호출 측에서 dict 구성 없이 바로 사용
# ─────────────────────────────────────────────────────────────────────────────
async def enqueue_trade_history(data: dict) -> None:
    """매매일지 INSERT 를 큐에 등록.

    [M-7 2026-10-02] payload 에 created_at 이 없으면 **큐 투입 시각**으로 채운다.
    이전에는 모델 기본값(now_kst)이 INSERT 시점에 적용돼, DLQ 재처리분은 created_at 이
    재처리 시각(다음 기동)이 됐다 → 당일 판정(`created_at >= 오늘`)·일별 집계가 어긋났다.
    호출자 dict 는 변경하지 않는다(복사).
    """
    if isinstance(data, dict) and data.get("created_at") is None:
        data = {**data, "created_at": now_kst()}
    await db_task_queue.put({"type": "insert_trade_history", "data": data})


async def enqueue_update_trade_history(trade_id: int, updates: dict) -> None:
    """기존 TradeHistory 레코드 UPDATE 를 큐에 등록.

    주로 체결 확인(verify_and_sync) 후 executed_price/quantity/status/slippage 반영 시 사용.

    Args:
        trade_id: 업데이트할 TradeHistory.id
        updates:  업데이트할 필드 dict (예: {"executed_quantity": 10, "status": "COMPLETED"})
    """
    await db_task_queue.put({
        "type": "update_trade_history",
        "data": {"id": trade_id, **updates},
    })


async def enqueue_upsert_analysis(row: dict) -> None:
    """단건 StockAnalysis UPSERT 를 큐에 등록."""
    await db_task_queue.put({"type": "upsert_stock_analysis", "data": row})


async def enqueue_bulk_upsert_analysis(rows: list[dict]) -> None:
    """다건 StockAnalysis batch UPSERT 를 큐에 등록."""
    if rows:
        await db_task_queue.put({"type": "bulk_upsert_stock_analysis", "data": rows})


async def enqueue_db_write(task: dict) -> None:
    """범용 DB 쓰기 태스크를 큐에 등록.

    task 는 반드시 'type' 키를 포함해야 하며, 워커가 지원하는 task type 중 하나여야 한다.
    예: {"type": "log_mcp_deny", "reason": "...", "trade_plan": {...}}
    """
    await db_task_queue.put(task)


async def enqueue_trade_journal_raw(data: dict) -> None:
    """TradeJournal INSERT 를 큐에 등록 (순수 dict 전달용)."""
    await db_task_queue.put({"type": "insert_trade_journal", "data": data})


async def enqueue_stock_ban_upsert(stock_code: str, consecutive_losses: int, ban_until) -> None:
    """StockBan UPSERT 를 큐에 등록 (loss_tracker에서 호출)."""
    await db_task_queue.put({
        "type": "upsert_stock_ban",
        "data": {
            "stock_code": stock_code,
            "consecutive_losses": consecutive_losses,
            "ban_until": ban_until,
        }
    })


async def enqueue_stock_ban_upsert_v2(
    user_id: int, stock_code: str, consecutive_losses: int, ban_until
) -> None:
    """[v2] StockBanV2 UPSERT 를 큐에 등록 (사용자 스코프 loss_tracker에서 호출).

    stock_bans(PK=stock_code, 단일 계정 전제)와 달리 stock_bans_v2 는
    UNIQUE(user_id, stock_code) 이므로 동일 종목을 사용자별로 독립 추적한다.
    """
    await db_task_queue.put({
        "type": "upsert_stock_ban_v2",
        "data": {
            "user_id": user_id,
            "stock_code": stock_code,
            "consecutive_losses": consecutive_losses,
            "ban_until": ban_until,
        }
    })


async def enqueue_portfolio_upsert(data: dict) -> None:
    """Portfolio INSERT/UPDATE 를 큐에 등록.

    data 필수 키: stock_code, quantity, average_price
    data 선택 키: stock_name, current_price, profit_loss, profit_loss_rate,
                  market, strategy_type ('swing' | 'daytrading')
    """
    await db_task_queue.put({"type": "upsert_portfolio", "data": data})


async def enqueue_active_trade_upsert(data: dict) -> None:
    """ActiveTrade UPSERT 를 큐에 등록.

    data 필수 키: stock_code, quantity, entry_price
    data 선택 키: strategy_name ('swing' | 'day_trade'), stop_loss,
                  take_profit, entry_reason, highest_since_entry
    data 선택 키(조회 필터 전용, ORM 컬럼 아님 — _ACTIVE_TRADE_META_KEYS):
                  strategy_names(list → IN),
                  strategy_name_not(str|list → NOT IN + strategy_name IS NULL 포함)
                  우선순위: strategy_name > strategy_names > strategy_name_not
    """
    await db_task_queue.put({"type": "upsert_active_trade", "data": data})


async def enqueue_close_active_trade(stock_code: str, strategy_name: str | None = None) -> None:
    """ActiveTrade 를 'closed' 상태로 변경하는 태스크를 큐에 등록.

    전략명 부정 조건(strategy_name_not) 등 확장 필터가 필요하면
    시그니처 변경 없이 enqueue_db_write() 로 payload 를 직접 넘긴다:
        await enqueue_db_write({
            "type": "close_active_trade",
            "data": {"stock_code": code, "strategy_name_not": DT_STRATEGY_NAMES},
        })
    """
    await db_task_queue.put({
        "type": "close_active_trade",
        "data": {"stock_code": stock_code, "strategy_name": strategy_name},
    })


async def enqueue_close_active_trade_v2(
    user_id: int, stock_code: str, strategy_name: str | None = None
) -> None:
    """[v2] 사용자 스코프 ActiveTrade 를 'closed' 상태로 변경하는 태스크를 큐에 등록.

    close_active_trade 와 동일 task type("close_active_trade")을 재사용한다 —
    핸들러가 data.get("user_id")로 필터를 추가하므로 새 타입이 필요 없다.
    """
    await db_task_queue.put({
        "type": "close_active_trade",
        "data": {"stock_code": stock_code, "strategy_name": strategy_name, "user_id": user_id},
    })


async def enqueue_update_trade_history_by_order_id(order_id: str, updates: dict) -> None:
    """KIS 주문번호(order_id)로 TradeHistory UPDATE 를 큐에 등록.

    trade_id 대신 order_id 로 레코드를 조회하여 업데이트한다.
    enqueue_trade_history() 로 INSERT 한 레코드의 id 를 모르는 경우 사용.

    Args:
        order_id: KIS 주문번호 (TradeHistory.order_id)
        updates:  업데이트할 필드 dict (예: {"executed_quantity": 10, "status": "COMPLETED"})
    """
    await db_task_queue.put({
        "type": "update_trade_history_by_order_id",
        "data": {"order_id": order_id, **updates},
    })


async def enqueue_trade_journal(trade, content: str) -> None:
    """TradeJournal INSERT 를 큐에 등록 (main.py post_market_scheduler 용).

    CLAUDE.md 3.2: 스케줄러(실시간 계층)에서 직접 DB 커밋 금지 → 큐 경유.
    """
    from datetime import datetime
    from app.services.trade_analyzer import _to_float, _parse_strategy, _parse_exit_type

    pl_rate = _to_float(trade.profit_loss_rate)
    # [2026-10-01] strategy_type 컬럼(swing/daytrading/manual)을 넘겨야 한다. 빠져 있어
    # entry_reason 키워드만 보다 16:00 배치 일지가 전부 'unknown' 이 됐다
    # (create_journal_record 는 넘기고 있었다 — 같은 거래가 경로에 따라 다른 전략명).
    strategy = _parse_strategy(trade.entry_reason, getattr(trade, "strategy_type", None))
    exit_type = _parse_exit_type(trade.exit_reason)
    mood = "positive" if pl_rate > 0 else ("neutral" if pl_rate > -3.0 else "negative")
    tags = [strategy, exit_type]
    if abs(pl_rate) > 5.0:
        tags.append("significant")

    await db_task_queue.put({
        "type": "insert_trade_journal",
        "data": {
            "trade_id": trade.id,
            "stock_code": trade.stock_code,
            "stock_name": trade.stock_name,
            "journal_date": trade.executed_at or trade.created_at or now_kst(),
            "content": content,
            "strategy_name": strategy,
            "mood": mood,
            "tags": tags,
        },
    })


# ─────────────────────────────────────────────────────────────────────────────
# 포지션 추적 계열 핸들러 (공통 종단)
#
# [F3 2026-09-28] 종료 drain 이 insert_trade_history 만 커밋하고
# upsert_portfolio / upsert_active_trade / close_active_trade 는 DLQ 로 보냈다
# → "체결 기록은 남는데 포지션 추적 행은 사라지는" 비대칭이 설계에 내장돼 있었다
# (재기동 시 restore_from_db 가 복원할 행이 없어 포지션이 고아가 된다).
# 정상 경로와 drain 이 **같은 구현**을 쓰도록 여기로 추출한다. 두 곳에 복제하면
# 한쪽만 바뀌어 더 나쁜 불일치가 된다(CLAUDE.md §3.3 과 동일한 취지).
# ─────────────────────────────────────────────────────────────────────────────


async def _handle_insert_trade_journal(session, data: dict) -> None:
    """TradeJournal INSERT — 정상 경로와 종료 drain 이 공유하는 단일 구현.

    [2026-09-29 B3] 주문 직후 백그라운드로 만드는 일지는 `TradeHistory.id` 를
    모르기 때문에 `trade_id=0` 을 넣고 있었다. `get_uncovered_trades()` 가
    `TradeJournal.trade_id` 로 커버 판정을 하므로 그 일지는 **아무 거래도 커버하지
    못하고**, 실제 TradeHistory 행이 영구 uncovered 로 남아 16:00 배치가 같은 매도에
    두 번째 일지를 만들었다(실측 중복 13건).

    호출부는 `order_id`(KIS 주문번호)를 알고 있으므로, 여기(이미 DB 세션 안 —
    CLAUDE.md §3 위반 없음)에서 `order_id → TradeHistory.id` 로 해소한다.
    해소 실패 시 `trade_id` 는 **None** 으로 남는다(0 이 아니다 — 0 은 존재하지 않는
    PK 를 가리키는 거짓 값이다).
    """
    from app.models.models import TradeJournal

    payload = dict(data)
    # order_id 는 TradeJournal 컬럼이 아니다 — 해소용 입력이므로 반드시 분리한다.
    order_id = payload.pop("order_id", None)

    if not payload.get("trade_id") and order_id:
        try:
            result = await session.execute(
                select(TradeHistory.id)
                .where(TradeHistory.order_id == str(order_id))
                .order_by(TradeHistory.id.desc())
                .limit(1)
            )
            resolved = result.scalar_one_or_none()
            if resolved:
                payload["trade_id"] = resolved
                # 해소된 trade_id 로 이미 일지가 있으면 중복 생성 금지 (장중 태스크·16:00 배치와
                # 즉시 생성 경로의 경합). 명시 trade_id 경로는 조회하지 않는다(기존 계약).
                dup = await session.execute(
                    select(TradeJournal.id).where(TradeJournal.trade_id == resolved).limit(1)
                )
                if dup.scalar_one_or_none() is not None:
                    logger.info(
                        "[Worker] TradeJournal skip — trade_id=%s 일지 이미 존재", resolved
                    )
                    return
            else:
                logger.warning(
                    "⚠️ [Worker] TradeJournal trade_id 미해소 (order_id=%s, %s) — "
                    "uncovered 판정에서 제외됨",
                    order_id, payload.get("stock_code", "?"),
                )
        except Exception as resolve_err:
            logger.warning(
                "⚠️ [Worker] trade_id 해소 실패 (order_id=%s): %s", order_id, resolve_err
            )

    # 0 은 과거 호출부의 플레이스홀더였다 — 거짓 커버 판정을 만들지 않도록 None 으로.
    if payload.get("trade_id") == 0:
        payload["trade_id"] = None

    session.add(TradeJournal(**payload))
    await session.commit()
    logger.debug(
        "✅ [Worker] TradeJournal saved: %s (trade_id=%s)",
        payload.get("stock_code", "?"), payload.get("trade_id"),
    )


async def _handle_upsert_portfolio(session, data: dict) -> None:
    """Portfolio UPSERT — 정상 경로와 종료 drain 이 공유하는 단일 구현."""
    from app.models.models import Portfolio

    stock_code = data.get("stock_code")
    user_id = data.get("user_id")
    query = select(Portfolio).where(Portfolio.stock_code == stock_code)
    if user_id is not None:
        query = query.where(Portfolio.user_id == user_id)
    result = await session.execute(query)
    record = result.scalar_one_or_none()

    if record is None:
        record = Portfolio(**data)
        session.add(record)
    else:
        for key, val in data.items():
            if key != "stock_code" and val is not None:
                setattr(record, key, val)

    await session.commit()
    logger.debug(
        "✅ [Worker] Portfolio upsert: %s (strategy=%s)",
        stock_code,
        data.get("strategy_type", "swing"),
    )


# [2026-10-01] update_trade_history_by_order_id 의 식별·필터 전용 키 — setattr 대상이 아니다.
# strategy_type/order_type 은 중복 행을 좁히는 데만 쓴다(좁히기에 실패해 다른 전략 행이
# 선택됐을 때 그 행의 전략 구분을 덮어쓰지 않도록).
_ORDER_ID_UPDATE_META_KEYS: frozenset = frozenset(
    {"order_id", "order_date", "strategy_type", "order_type"}
)


def _norm_kis_order_id(value) -> str:
    """KIS 주문번호 정규화(앞자리 0 제거). 저장 형식은 응답 ODNO 원문(0 패딩)이다."""
    s = str(value or "").strip()
    return s.lstrip("0")


def _order_id_ref_date(data: dict) -> date:
    """조회 기준일(KST). 우선순위: order_date(메타) > created_at > executed_at > 오늘.

    executed_at 은 체결 확인 시각(주문 후 수 초)이라 주문일과 같다. DLQ 재처리로 다음 날
    실행돼도 payload 의 원래 날짜를 쓰므로 당일 필터가 행을 놓치지 않는다.
    """
    for key in ("order_date", "created_at", "executed_at"):
        v = data.get(key)
        if v is None or v == "":
            continue
        try:
            if isinstance(v, datetime):
                return v.date()
            if isinstance(v, date):
                return v
            if isinstance(v, str):
                return datetime.fromisoformat(v).date() if "T" in v or " " in v else date.fromisoformat(v)
        except (TypeError, ValueError):
            logger.warning("⚠️ [Worker] order_id 갱신 기준일 파싱 실패 %s=%r — 다음 후보 사용", key, v)
    return now_kst().date()


async def _handle_update_trade_history_by_order_id(session, data: dict) -> None:
    """TradeHistory UPDATE by KIS 주문번호 — 정상 경로 단일 구현.

    [2026-10-01] KIS ODNO 는 **일별 순번**이라 날짜가 바뀌면 같은 번호가 반복된다. 예전 구현
    (`order_id == X` + `scalar_one_or_none()`)은 과거 행과 겹치면 MultipleResultsFound →
    단일 워커가 1+2+4초 재시도로 큐 전체를 멈추고 DLQ 로 보냈고, 당일 INSERT 가 유실된
    상태면 **과거 행을 덮어써** 원장을 오염시켰다.
      - 기준일(KST 00:00~다음날 00:00) created_at 범위로 한정한다(_order_id_ref_date).
      - order_id 는 원문 일치 OR 앞자리 0 제거 일치.
      - 그래도 여러 행이면 payload 의 strategy_type·order_type 으로 좁히고 created_at desc
        첫 행만 갱신 + warning (예외 없음 → 큐 정체 없음).
      - 0건이면 warning 후 종료(재시도·DLQ 없음 — 기존 동작 유지).
    """
    from datetime import timedelta
    from sqlalchemy import func

    order_id = data.get("order_id")
    if not order_id:
        return
    user_id = data.get("user_id")
    ref = _order_id_ref_date(data)
    start = datetime.combine(ref, datetime.min.time())
    end = start + timedelta(days=1)

    raw = str(order_id).strip()
    norm = _norm_kis_order_id(raw)
    id_cond = TradeHistory.order_id == raw
    if norm:
        id_cond = or_(id_cond, func.ltrim(TradeHistory.order_id, "0") == norm)

    query = select(TradeHistory).where(
        id_cond,
        TradeHistory.created_at >= start,
        TradeHistory.created_at < end,
    )
    if user_id is not None:
        query = query.where(TradeHistory.user_id == user_id)
    query = query.order_by(TradeHistory.created_at.desc(), TradeHistory.id.desc())
    rows = list((await session.execute(query)).scalars().all())

    if not rows:
        logger.warning(
            "⚠️ [Worker] TradeHistory not found for order_id=%s (기준일 %s KST) — 갱신 생략",
            order_id, ref.isoformat(),
        )
        return

    if len(rows) > 1:
        narrowed = rows
        for key in ("strategy_type", "order_type"):
            want = data.get(key)
            if want is None:
                continue
            cand = [r for r in narrowed if getattr(r, key, None) == want]
            if cand:
                narrowed = cand
        logger.warning(
            "⚠️ [Worker] order_id=%s 기준일 %s 에 TradeHistory %d건 중복 — "
            "strategy_type/order_type 으로 좁힌 %d건 중 최신 1건(id=%s)만 갱신",
            order_id, ref.isoformat(), len(rows), len(narrowed), narrowed[0].id,
        )
        rows = narrowed

    trade = rows[0]
    for key, val in data.items():
        if key in _ORDER_ID_UPDATE_META_KEYS or val is None or not hasattr(trade, key):
            continue
        setattr(trade, key, val)
    await session.commit()
    logger.info(
        "✅ [Worker] TradeHistory updated by order_id=%s (id=%s) status=%s",
        order_id, trade.id, data.get("status", "?"),
    )


def _apply_entry_price_adjustments_v2(record, data: dict) -> frozenset:
    """[2026-10-08] 기존 active 행의 진입가·수량을 '누적 원가' 기준으로 갱신한다.

    - ``accumulate_entry=True`` (추가매수): quantity = 기존 + 신규, entry_price = 수량 가중평균,
      highest_since_entry = max(기존, 신규). 이전에는 신규 매수 수량·주문가로 **덮어써**
      스윙 진입가가 마지막 추가매수가로 바뀌었다(스윙 손익·손절선 왜곡).
    - ``entry_fill_adjust={"order_price","filled_price","filled_qty"}`` (체결 확인 후):
      entry_price += (체결가 − 주문가) × 체결수량 / 행 수량. 신규 진입(행 수량 == 체결수량)이면
      정확히 체결가가 된다. 추가매수였다면 그 매수분의 원가 차이만 반영된다.

    둘 다 비멱등이라 계산한 절대값을 payload 에 고정한다(`_applied_*` + `_target_id`,
    remaining_pct_multiply 와 같은 규약). 정상 재시도에서는 `_strip_retry_markers` 가 지운다.

    Returns:
        일반 setattr 루프에서 건너뛸 키 집합.
    """
    skip: set = set()
    _rid = getattr(record, "id", None)
    _pinned = (
        data.get("_applied_entry_price") is not None
        and _rid is not None
        and data.get("_target_id") == _rid
    )

    if data.get("accumulate_entry"):
        # [2026-10-09] remaining_pct 도 함께 재계산한다(불변식: 실보유 = quantity × remaining_pct).
        # 호출부가 넘긴 remaining_pct 는 무시한다(옛 스냅샷 값이 재계산을 덮어쓰는 것을 방지).
        skip.update({"quantity", "entry_price", "highest_since_entry", "remaining_pct"})
        if _pinned:
            record.entry_price = float(data["_applied_entry_price"])
            if data.get("_applied_quantity") is not None:
                record.quantity = int(data["_applied_quantity"])
            if data.get("_applied_remaining_pct") is not None:
                record.remaining_pct = float(data["_applied_remaining_pct"])
        else:
            old_q = int(getattr(record, "quantity", 0) or 0)
            old_p = float(getattr(record, "entry_price", 0) or 0)
            add_q = int(data.get("quantity") or 0)
            add_p = float(data.get("entry_price") or 0)
            new_q = old_q + add_q
            # 가중치는 '현재 보유분' = 원본 수량 × remaining_pct (분할 익절 후 추가매수 시
            # 이미 판 물량의 원가가 평단에 섞이지 않게). quantity 컬럼은 원본 누적 규약 유지.
            try:
                _rp = float(getattr(record, "remaining_pct", 1.0) or 1.0)
            except (TypeError, ValueError):
                _rp = 1.0
            _rp = min(1.0, max(0.0, _rp))
            held_q = old_q * _rp
            if add_q > 0 and add_p > 0 and held_q > 0 and old_p > 0:
                new_p = (held_q * old_p + add_q * add_p) / (held_q + add_q)
            elif add_q > 0 and add_p > 0:
                new_p = add_p
            else:
                new_p = old_p
            record.quantity = new_q
            record.entry_price = round(float(new_p), 2)  # Numeric(15,2)
            # 분할 익절 뒤 추가매수: 원본 누적(quantity)이 늘었으므로 remaining_pct 를
            # (현재 보유 + 추가) / (원본 + 추가) 로 맞춘다. 안 맞추면 실보유/remaining_pct 로
            # 역산한 원본이 부풀어 다음 분할매도 수량이 과대(과다매도)가 된다.
            _rp_changed = add_q > 0 and new_q > 0 and _rp < 1.0
            if _rp_changed:
                record.remaining_pct = round(min(1.0, (held_q + add_q) / new_q), 6)
            if _rid is not None:
                data["_applied_entry_price"] = float(record.entry_price)
                data["_applied_quantity"] = int(record.quantity)
                if _rp_changed:
                    data["_applied_remaining_pct"] = float(record.remaining_pct)
                data["_target_id"] = _rid
        _new_high = data.get("highest_since_entry")
        if _new_high is not None:
            _old_high = float(getattr(record, "highest_since_entry", 0) or 0)
            record.highest_since_entry = max(_old_high, float(_new_high))
        return frozenset(skip)

    adj = data.get("entry_fill_adjust")
    if isinstance(adj, dict):
        skip.add("entry_price")
        if _pinned:
            record.entry_price = float(data["_applied_entry_price"])
            return frozenset(skip)
        try:
            o = float(adj.get("order_price") or 0)
            f = float(adj.get("filled_price") or 0)
            q = int(adj.get("filled_qty") or 0)
        except (TypeError, ValueError):
            logger.warning("⚠️ [Worker] entry_fill_adjust 파싱 실패 — 진입가 보정 생략: %r", adj)
            return frozenset(skip)
        rec_q = int(getattr(record, "quantity", 0) or 0)
        cur = float(getattr(record, "entry_price", 0) or 0)
        try:
            _rp2 = min(1.0, max(0.0, float(getattr(record, "remaining_pct", 1.0) or 1.0)))
        except (TypeError, ValueError):
            _rp2 = 1.0
        # 평단은 '현재 보유분' 가중이므로 분모도 실보유(원본 × remaining_pct)로 맞춘다.
        held_q2 = rec_q * _rp2
        if o > 0 and f > 0 and q > 0 and held_q2 >= 1 and cur > 0:
            q_eff = min(q, held_q2)
            new_p = cur + (f - o) * q_eff / held_q2
            if new_p > 0:
                record.entry_price = round(float(new_p), 2)  # Numeric(15,2)
                if _rid is not None:
                    data["_applied_entry_price"] = float(record.entry_price)
                    data["_target_id"] = _rid
        else:
            logger.warning(
                "⚠️ [Worker] ActiveTrade 체결가 보정 조건 불충족 — 생략 "
                "(order=%s filled=%s qty=%s row_qty=%s row_entry=%s)", o, f, q, rec_q, cur,
            )
    return frozenset(skip)


async def _handle_upsert_active_trade(session, data: dict) -> None:
    """ActiveTrade UPSERT — 정상 경로와 종료 drain 이 공유하는 단일 구현."""
    from app.models.models import ActiveTrade

    stock_code = data.get("stock_code")
    strategy_name = data.get("strategy_name")
    user_id = data.get("user_id")

    query = select(ActiveTrade).where(
        ActiveTrade.stock_code == stock_code,
        ActiveTrade.status == "active",
    )
    # [§8.5 항목6] strategy_name(동등) / strategy_names(IN) /
    # strategy_name_not(NOT IN + NULL 포함) — 새 키 미사용 시 무변경.
    query = _apply_strategy_name_filter(query, data, ActiveTrade.strategy_name)
    if user_id is not None:
        query = query.where(ActiveTrade.user_id == user_id)
        # [2026-09-27 §8.5 후속] legacy 경로와 동일하게 '가장 최근 행' 우선.
        # scalar_one_or_none() 은 한 사용자가 같은 종목에 비-DT active 행을
        # 2건 이상 보유하면 MultipleResultsFound → 3회 재시도 후 DLQ 로 빠진다.
        # (멀티테넌트 플래그 ON 전제조건)
        query = query.order_by(ActiveTrade.id.desc()).limit(1)
        result = await session.execute(query)
        record = result.scalars().first()
    else:
        # legacy(user_id=None) 경로: 사용자 다중화로 동일 종목 active
        # 행이 2건 이상 존재할 수 있어 scalar_one_or_none() 이
        # MultipleResultsFound 를 던질 수 있음 → 가장 최근 행 우선.
        query = query.order_by(ActiveTrade.id.desc()).limit(1)
        result = await session.execute(query)
        record = result.scalars().first()

    if record is None and data.get("entry_fill_adjust"):
        # 체결가 보정 전용 payload(수량·진입가 없음) — 대상 행이 없으면(이미 청산 등) 할 일이 없다.
        logger.info(
            "ℹ️ [Worker] ActiveTrade 체결가 보정 생략: %s active 행 없음 (strategy=%s)",
            stock_code, strategy_name,
        )
        return

    if record is None:
        # remaining_pct_multiply·전략명 필터 키는 신규 생성 시 무시
        create_data = {
            k: v for k, v in data.items()
            if k not in _ACTIVE_TRADE_META_KEYS
        }
        # 필수 컬럼(quantity/entry_price) 없이 신규 생성 요청이 오면
        # IntegrityError로 3회 재시도만 낭비 → 경고 후 스킵
        _missing = [k for k in ("quantity", "entry_price") if create_data.get(k) is None]
        if _missing:
            logger.warning(
                "⚠️ [Worker] ActiveTrade upsert 스킵: %s active 행 없음 + 필수 키 누락 %s (keys=%s)",
                stock_code, _missing, sorted(create_data.keys()),
            )
            return
        if data.get("_retried_exec"):
            logger.warning(
                "⚠️ [Worker] ActiveTrade 신규 생성이 재시도 실행에서 발생: %s (strategy=%s) — "
                "순서 역전/유령 행 가능성, KIS 잔고와 대조 필요",
                stock_code, strategy_name,
            )
        record = ActiveTrade(**create_data)
        session.add(record)
    else:
        # remaining_pct_multiply: 기존 remaining_pct에 비율을 곱함
        # (MDD 축소 시 기존 분할매도 이력 반영)
        # data.get() 사용 — pop()으로 task dict를 파괴적으로 변경하면
        # 재시도 시 이 키가 사라져 no-op으로 성공 처리되는 버그 방지
        multiply_ratio = data.get("remaining_pct_multiply")
        if multiply_ratio is not None:
            # [M-A 2026-10-02] 곱셈은 비멱등이다 — 같은 실행이 계산한 절대값을 payload 에
            # 고정해, 커밋 여부 불명으로 재실행(종료 drain)돼도 두 번 곱하지 않는다.
            # 정상 재시도에서는 이 마커를 지운다(_strip_retry_markers).
            _applied = data.get("_applied_remaining_pct")
            _rid = getattr(record, "id", None)
            if _applied is not None and _rid is not None and data.get("_target_id") == _rid:
                record.remaining_pct = float(_applied)
            else:
                current_remaining = getattr(record, "remaining_pct", 1.0) or 1.0
                record.remaining_pct = float(current_remaining) * float(multiply_ratio)
                if _rid is not None:
                    data["_applied_remaining_pct"] = record.remaining_pct
                    data["_target_id"] = _rid
        # [2026-10-08] 추가매수(accumulate_entry) / 체결가 보정(entry_fill_adjust) 은 진입가·수량을
        # **기존 행 기준으로 계산**해야 하므로 아래 일반 setattr 루프에서 해당 키를 제외한다.
        _skip_keys = _apply_entry_price_adjustments_v2(record, data)
        for key, val in data.items():
            if key in _skip_keys:
                continue
            # [C1 2026-09-28] 반드시 continue — return 이면 'stock_code' 가 모든 payload 의
            # 첫 키라서 기존 active 행에 대한 모든 upsert 가 첫 이터레이션에서 빠져나가
            # commit() 에 도달하지 못한다(예외가 없어 성공 처리되고 DLQ 에도 안 남는다).
            # 파급: 분할매도 잔량 갱신·remaining_pct_multiply·highest_since_entry 트레일링
            # 갱신이 전부 무효 → 2026-09-23 '4주 보유 중 7주 매도' 재발 경로.
            if key == "stock_code" or key in _ACTIVE_TRADE_META_KEYS:
                continue
            if val is not None:
                setattr(record, key, val)

    await session.commit()
    logger.debug(
        "✅ [Worker] ActiveTrade upsert: %s (strategy=%s)",
        stock_code,
        strategy_name,
    )


async def _handle_upsert_stock_ban(session, data: dict) -> None:
    """StockBan UPSERT — 정상 경로와 종료 drain 이 공유 (M-3 2026-10-02: 멱등이라 drain 커밋 대상)."""
    from app.models.models import StockBan
    from app.core.dt_utils import to_naive_utc

    stock_code = data.get("stock_code")
    consecutive_losses = data.get("consecutive_losses")
    # stock_bans.ban_until은 tz-naive 컬럼 — tz-aware 값이 들어와도 안전하게 저장.
    # [시각 규약 — 의도적] ban_until 은 naive-**UTC** (loss_tracker 가 to_aware_utc 로 읽는다).
    # 같은 행의 updated_at 등 다른 테이블 시각은 KST-naive 라 SQL 에서 `ban_until > localtimestamp`
    # 처럼 직접 비교하면 9시간 어긋난다 -> `(ban_until AT TIME ZONE 'UTC') > now()` 를 쓸 것.
    # 저장만 KST 로 바꾸면 기존 활성 밴이 9시간 지연 해제되고(load 도 같이 바꿔야 함) 마이그레이션이
    # 필요하므로 규약을 유지한다(테스트로 고정: test_db_worker_stock_ban_tz_20261010).
    ban_until = to_naive_utc(data.get("ban_until"))

    result = await session.execute(select(StockBan).where(StockBan.stock_code == stock_code))
    record = result.scalar_one_or_none()

    if record is None:
        record = StockBan(
            stock_code=stock_code,
            consecutive_losses=consecutive_losses,
            ban_until=ban_until,
        )
        session.add(record)
    else:
        record.consecutive_losses = consecutive_losses
        record.ban_until = ban_until

    await session.commit()
    logger.debug(
        "✅ [Worker] StockBan upsert: %s (losses=%d, ban_until=%s)",
        stock_code,
        consecutive_losses,
        ban_until.isoformat() if ban_until else "None",
    )


async def _handle_upsert_stock_ban_v2(session, data: dict) -> None:
    """StockBanV2 UPSERT (ON CONFLICT DO UPDATE) — 정상 경로와 종료 drain 이 공유."""
    from app.models.models import StockBanV2
    from sqlalchemy.dialects.postgresql import insert as pg_insert
    from app.core.dt_utils import to_naive_utc

    user_id = data.get("user_id")
    stock_code = data.get("stock_code")
    consecutive_losses = data.get("consecutive_losses")
    # stock_bans_v2.ban_until도 tz-naive 컬럼 — tz-aware 값이 들어와도 안전하게 저장.
    ban_until = to_naive_utc(data.get("ban_until"))

    # PostgreSQL INSERT ... ON CONFLICT DO UPDATE — 여러 워커/사용자
    # 사이클이 동시에 같은 (user_id, stock_code)를 갱신해도 경쟁 안전.
    stmt = pg_insert(StockBanV2).values(
        user_id=user_id,
        stock_code=stock_code,
        consecutive_losses=consecutive_losses,
        ban_until=ban_until,
    )
    stmt = stmt.on_conflict_do_update(
        constraint="uq_stock_bans_v2_user_stock",
        set_={
            "consecutive_losses": stmt.excluded.consecutive_losses,
            "ban_until": stmt.excluded.ban_until,
            "updated_at": now_kst(),
        },
    )
    await session.execute(stmt)
    await session.commit()
    logger.debug(
        "✅ [Worker] StockBanV2 upsert: user=%s %s (losses=%s, ban_until=%s)",
        user_id, stock_code, consecutive_losses,
        ban_until.isoformat() if ban_until else "None",
    )


async def _handle_close_active_trade(session, data: dict) -> None:
    """ActiveTrade close — 정상 경로와 종료 drain 이 공유하는 단일 구현."""
    from app.models.models import ActiveTrade
    from datetime import datetime as _dt

    stock_code = data.get("stock_code")
    strategy_name = data.get("strategy_name")
    user_id = data.get("user_id")

    # [M-A 2026-10-02] 멱등화: 한 번 대상 행을 고르면 payload 에 그 id 를 고정한다.
    # "가장 최근 active 행을 닫아라"는 재실행할 때마다 **다음** 행을 닫는다 — 커밋 직후
    # cancel 된 태스크를 종료 drain 이 다시 돌리면 같은 종목의 다른 포지션(DT↔스윙)까지
    # 닫혔다(9/23 유령 행 계열). 고정된 id 가 이미 closed 면 no-op 이다.
    pinned_id = data.get("_target_id")
    if pinned_id is not None:
        if pinned_id < 0:  # 처음 실행 때 대상 없음 → 재실행도 no-op
            return
        result = await session.execute(
            select(ActiveTrade).where(
                ActiveTrade.id == pinned_id, ActiveTrade.status == "active"
            )
        )
        record = result.scalars().first()
    else:
        query = select(ActiveTrade).where(
            ActiveTrade.stock_code == stock_code,
            ActiveTrade.status == "active",
        )
        # [§8.5 항목6] strategy_name(동등) / strategy_names(IN) /
        # strategy_name_not(NOT IN + NULL 포함) — 새 키 미사용 시 무변경.
        query = _apply_strategy_name_filter(query, data, ActiveTrade.strategy_name)
        if user_id is not None:
            query = query.where(ActiveTrade.user_id == user_id)
        # [2026-09-27 §8.5 후속] '가장 최근 행' 우선 — scalar_one_or_none() 은 같은 종목
        # active 행이 2건 이상이면 MultipleResultsFound 로 3회 재시도 후 DLQ 로 빠진다.
        query = query.order_by(ActiveTrade.id.desc()).limit(1)
        result = await session.execute(query)
        record = result.scalars().first()
        # id 를 알 수 없으면(None) 고정하지 않는다 — 기존 동작으로 폴백
        data["_target_id"] = getattr(record, "id", None) if record is not None else -1

    if record:
        record.status = "closed"
        record.updated_at = now_kst()
        await session.commit()
        logger.debug(
            "✅ [Worker] ActiveTrade closed: %s (strategy=%s)",
            stock_code,
            strategy_name,
        )
    else:
        logger.warning(
            "⚠️ [Worker] ActiveTrade close 대상 0건: %s (strategy=%s, pinned=%s, retried=%s)",
            stock_code, strategy_name, pinned_id, bool(data.get("_retried_exec")),
        )


# ─────────────────────────────────────────────────────────────────────────────
# 워커 본체
# ─────────────────────────────────────────────────────────────────────────────
async def _execute_db_task_impl(task: dict) -> None:
    """[H-5 2026-10-02] 단일 태스크 실행 (워커 본체에서 분리 — 로직 무변경).

    예외는 호출자(db_writer_worker)가 재시도/DLQ 로 처리한다.
    """
    task_type = task.get("type")
    # [M-A] 멱등화 마커를 payload 에 쓰는 타입은 호출자 dict 와 분리한다(최초 1회 복사).
    if task_type in _PINNED_TASK_TYPES and isinstance(task.get("data"), dict) and not task.get("_data_owned"):
        task["data"] = dict(task["data"])
        task["_data_owned"] = True
    data = task.get("data")
    if task_type in _PINNED_TASK_TYPES and isinstance(data, dict) and task.get("_requeued"):
        data["_retried_exec"] = True

    async with AsyncSessionLocal() as session:
        # ── 1. 매매일지 INSERT ─────────────────────────────────────
        if task_type == "insert_trade_history":
            session.add(TradeHistory(**data))
            await session.commit()
            logger.info(
                "✅ [Worker] TradeHistory saved: %s",
                data.get("stock_code", "?"),
            )

        # ── 2. TradeHistory UPDATE (체결 확인 후 슬리피지/수량 반영) ──
        elif task_type == "update_trade_history":
            trade_id = data.get("id")
            user_id = data.get("user_id")
            if trade_id:
                query = select(TradeHistory).where(TradeHistory.id == trade_id)
                if user_id is not None:
                    query = query.where(TradeHistory.user_id == user_id)
                row = await session.execute(query)
                trade = row.scalar_one_or_none()
                if trade:
                    for key, val in data.items():
                        if key != "id" and val is not None and hasattr(trade, key):
                            setattr(trade, key, val)
                    await session.commit()
                    logger.info(
                        "✅ [Worker] TradeHistory updated: id=%s status=%s",
                        trade_id,
                        data.get("status", "?"),
                    )

        # ── 2b. TradeHistory UPDATE by order_id (KIS 주문번호 기준) ──
        elif task_type == "update_trade_history_by_order_id":
            await _handle_update_trade_history_by_order_id(session, data)

        # ── 3. 단건 StockAnalysis UPSERT ──────────────────────────
        elif task_type == "upsert_stock_analysis":
            from app.services.analysis_db_service import _execute_upsert
            await _execute_upsert(session, [data])
            await session.commit()
            logger.debug(
                "✅ [Worker] StockAnalysis upsert: %s",
                data.get("stock_code", "?"),
            )

        # ── 3. 다건 StockAnalysis batch UPSERT ────────────────────
        elif task_type == "bulk_upsert_stock_analysis":
            from app.services.analysis_db_service import _execute_upsert
            await _execute_upsert(session, data)          # data = list[dict]
            await session.commit()
            logger.info(
                "✅ [Worker] Bulk StockAnalysis upsert: %d rows", len(data)
            )

        # ── 4. StockBan UPSERT ────────────────────────────────────────
        elif task_type == "upsert_stock_ban":
            await _handle_upsert_stock_ban(session, data)

        # ── 4v2. StockBanV2 UPSERT (user_id, stock_code 스코프) ────────
        elif task_type == "upsert_stock_ban_v2":
            await _handle_upsert_stock_ban_v2(session, data)

        # ── 4b. News INSERT (중복 URL 무시) ────────────────────────────
        elif task_type == "insert_news":
            from app.models.models import News

            news_url = data.get("url")
            is_dup = False
            # URL 기반 중복 방지 (동일 URL은 skip)
            if news_url:
                dup_result = await session.execute(
                    select(News.id).where(News.url == news_url).limit(1)
                )
                if dup_result.scalar_one_or_none() is not None:
                    is_dup = True
                    logger.debug(
                        "⏭️ [Worker] News skip (dup URL): %s",
                        (data.get("title") or "?")[:40],
                    )

            if not is_dup:
                session.add(News(**data))
                await session.commit()
                logger.debug(
                    "✅ [Worker] News saved: %s",
                    (data.get("title") or "?")[:40],
                )

        # ── 5. MCP 거부 이벤트 로그 ───────────────────────────────────
        elif task_type == "log_mcp_deny":
            # DB 저장 없이 로그만 기록 (필요 시 별도 테이블 확장 가능)
            logger.warning(
                "🛑 [Worker] MCP DENY logged: reason=%s, trade_plan=%s",
                task.get("reason", "unknown"),
                task.get("trade_plan"),
            )

        # ── 6. Portfolio UPSERT ────────────────────────────────────────
        elif task_type == "upsert_portfolio":
            await _handle_upsert_portfolio(session, data)

        # ── 7. ActiveTrade UPSERT ──────────────────────────────────────
        elif task_type == "upsert_active_trade":
            await _handle_upsert_active_trade(session, data)

        # ── 8. ActiveTrade close ───────────────────────────────────────
        elif task_type == "close_active_trade":
            await _handle_close_active_trade(session, data)

        # ── 9. TradeJournal INSERT ────────────────────────────────────
        elif task_type == "insert_trade_journal":
            await _handle_insert_trade_journal(session, data)

        else:
            logger.warning("⚠️  [Worker] Unknown task type: %s", task_type)



# 처리 중 cancel 됐을 때 **커밋 여부를 알 수 없는** 비멱등 INSERT 계열.
# 종료 drain 에서 다시 커밋하면 중복 행이 생길 수 있어 DLQ 로 보관한다.
# replay_dlq 는 이 사유(_DLQ_COMMIT_UNKNOWN_REASON)를 **검증 후에만** 재큐잉한다(M-B).
# close_active_trade / upsert_active_trade(multiply) 는 핸들러가 대상 id·절대값을 payload 에
# 고정하므로(M-A) 재실행이 멱등이다 → 여기 넣지 않고 drain 에서 재실행한다.
_DLQ_COMMIT_UNKNOWN_REASON = "shutdown_inflight_cancelled_commit_unknown"
_NON_IDEMPOTENT_TASK_TYPES: tuple[str, ...] = (
    "insert_trade_history",
    "insert_trade_journal",
    "insert_news",
    # [L4] 로그만 남기는 타입 — 재실행하면 같은 거부 이벤트가 두 번 기록된다
    "log_mcp_deny",
)


# 종료 drain 에서 커밋까지 시도하는 타입 (나머지는 세션을 열지 않고 바로 DLQ)
_DRAIN_COMMIT_TYPES: tuple[str, ...] = (
    "insert_trade_history",
    "upsert_portfolio",
    "upsert_active_trade",
    "close_active_trade",
    "insert_trade_journal",
    # [M-3 2026-10-02] 연패 밴은 **재생성되지 않는다**(3연패 이력은 다시 계산되지 않음). 이전에는
    # '분석 계열'로 분류돼 DLQ 직행 → 3연패 직후 배포하면 42일 밴이 사라질 수 있었다. 멱등 upsert.
    "upsert_stock_ban",
    "upsert_stock_ban_v2",
)
# 분석 계열은 다음 배치/기동 시 재생성되므로 커밋하지 않고 DLQ 보관 (F4 replay_dlq 가 재큐잉)
_DRAIN_DLQ_ONLY_TYPES: tuple[str, ...] = (
    "update_trade_history",
    "update_trade_history_by_order_id",
    "upsert_stock_analysis",
    "bulk_upsert_stock_analysis",
)


async def _drain_commit_impl(drain_type: str, drain_data) -> None:
    async with AsyncSessionLocal() as drain_session:
        if drain_type == "insert_trade_history":
            drain_session.add(TradeHistory(**drain_data))
            await drain_session.commit()
        # [F3 2026-09-28] 포지션 추적 계열도 커밋한다.
        # 이전에는 insert_trade_history 만 커밋하고 아래 3종을 DLQ 로
        # 보내서 "체결 기록은 남는데 포지션 추적 행은 사라지는" 비대칭이
        # 있었다 → 재기동 시 restore_from_db 가 복원할 행이 없어
        # 포지션이 고아가 되고 손절·EOD 청산이 적용되지 않았다.
        elif drain_type == "upsert_portfolio":
            await _handle_upsert_portfolio(drain_session, drain_data)
        elif drain_type == "upsert_active_trade":
            await _handle_upsert_active_trade(drain_session, drain_data)
        elif drain_type == "close_active_trade":
            await _handle_close_active_trade(drain_session, drain_data)
        elif drain_type == "insert_trade_journal":
            await _handle_insert_trade_journal(drain_session, drain_data)
        elif drain_type == "upsert_stock_ban":
            await _handle_upsert_stock_ban(drain_session, drain_data)
        elif drain_type == "upsert_stock_ban_v2":
            await _handle_upsert_stock_ban_v2(drain_session, drain_data)


async def _drain_one_impl(drain_task: dict, drain_started: float) -> None:
    """종료 drain 1건 처리 — 커밋 또는 DLQ. 예외를 밖으로 내지 않는다.

    [M1 2026-10-02] 예산을 항목 **시작 전에만** 확인하면, DB 다운 시 세션 열기(asyncpg connect
    timeout ≈60초)가 항목 하나로 예산을 넘겨 stop_grace 를 초과 → SIGKILL 로 남은 큐가 DLQ 에도
    남지 않았다. 이제 세션 열기+커밋 전체를 **남은 예산**으로 `wait_for` 한다.
    """
    drain_type = drain_task.get("type", "unknown") if isinstance(drain_task, dict) else "unknown"
    try:
        drain_data = drain_task.get("data")
        # [F3 2026-09-28] 시간 예산 초과분은 커밋을 시도하지 않고 DLQ 로 보낸다.
        budget = DRAIN_DEADLINE_SECONDS - (time.monotonic() - drain_started)
        if budget <= 0:
            logger.error(
                "❌ [DBWorker] Drain 시간 예산 %.0f초 초과 — DLQ 로 보관 (type=%s)",
                DRAIN_DEADLINE_SECONDS, drain_type,
            )
            await _save_to_dlq(drain_task, "shutdown_drain_deadline_exceeded")
            return
        if drain_type in _DRAIN_DLQ_ONLY_TYPES:
            logger.error(
                "❌ [DBWorker] Drain 미커밋 → DLQ (type=%s) — 기동 시 replay_dlq() 로 재처리",
                drain_type,
            )
            await _save_to_dlq(drain_task, "shutdown_drain")
            return
        if drain_type not in _DRAIN_COMMIT_TYPES:
            logger.error("❌ [DBWorker] Drain 미지원 type → DLQ (type=%s)", drain_type)
            await _save_to_dlq(drain_task, "shutdown_drain_unknown_type")
            return
        try:
            await asyncio.wait_for(_drain_commit_impl(drain_type, drain_data), timeout=budget)
        except asyncio.TimeoutError:
            # 커밋 도중 끊겼을 수 있다 — 비멱등 INSERT 는 '커밋 여부 불명' 사유로 보관해
            # replay_dlq 가 중복 확인 후에만 재큐잉하게 한다(M-B).
            _reason = (
                _DLQ_COMMIT_UNKNOWN_REASON if drain_type in _NON_IDEMPOTENT_TASK_TYPES
                else "shutdown_drain_deadline_exceeded"
            )
            logger.error(
                "❌ [DBWorker] Drain 커밋이 남은 예산 %.1f초 안에 끝나지 않음 — DLQ 로 보관 (type=%s, 사유=%s)",
                budget, drain_type, _reason,
            )
            await _save_to_dlq(drain_task, _reason)
    except Exception as drain_err:
        logger.error(f"[DBWorker] Drain 실패: {drain_err}")
        try:
            await _save_to_dlq(drain_task, str(drain_err))
        except Exception as _swallowed:
            logger.warning("[DBWorker] Drain 실패 건의 DLQ 저장도 실패 — 태스크 유실 가능: %s", _swallowed)


async def _drain_on_shutdown_impl(
    queue: asyncio.Queue,
    inflight: dict | None,
    backoff: list[dict],
    pending_saves: set,
    drain_started: float | None = None,
) -> int:
    """[H-5] 종료 drain. 처리 순서: 진행 중 DLQ 저장 완료 대기 → in-flight → 백오프 대기분 → 큐 잔여.

    - in-flight: 처리 중 cancel 됐다. 비멱등 INSERT 는 커밋 여부를 모르므로 DLQ, 나머지는 재실행.
    - 백오프 대기분: 실패 후 재시도 대기 중이었다(큐 밖) — 이전에는 drain 대상에서 빠져 유실됐다.
    - 이 둘은 이미 queue.task_done() 이 호출됐으므로 여기서 다시 부르지 않는다.
    - drain_started: 같은 종료에서 두 번째 호출이면 첫 호출의 시작 시각을 넘긴다(예산 공유, M1).
    """
    if drain_started is None:
        drain_started = time.monotonic()
    remaining = 0

    # 0) 백오프 전에 시작된 DLQ 저장(최대 재시도 초과분)이 끝나기를 기다린다
    for fut in list(pending_saves):
        try:
            budget = max(0.0, DRAIN_DEADLINE_SECONDS - (time.monotonic() - drain_started))
            await asyncio.wait_for(asyncio.shield(fut), timeout=budget)
        except Exception as e:
            logger.error("❌ [DBWorker] 종료 중 DLQ 저장 대기 실패: %s", e)

    # 1) in-flight
    if inflight is not None:
        _itype = inflight.get("type", "unknown")
        if _itype in _NON_IDEMPOTENT_TASK_TYPES:
            logger.error(
                "❌ [DBWorker] 처리 중 종료 — 커밋 여부 불명 → DLQ 보관 (type=%s, 중복 가능성 확인 필요)",
                _itype,
            )
            try:
                await _save_to_dlq(inflight, _DLQ_COMMIT_UNKNOWN_REASON)
            except Exception as _swallowed:
                logger.warning("[DBWorker] in-flight DLQ 저장 실패 — 태스크 유실 가능: %s", _swallowed)
        else:
            await _drain_one_impl(inflight, drain_started)
        remaining += 1

    # 2) 백오프 대기분 (실패 이력이 있어 아직 커밋되지 않았음이 확실하다)
    for t in list(backoff):
        await _drain_one_impl(t, drain_started)
        remaining += 1

    # 3) 큐 잔여
    while not queue.empty():
        try:
            drain_task = queue.get_nowait()
        except asyncio.QueueEmpty:
            # get_nowait()이 실패해 아이템을 꺼내지 못했으므로 task_done() 호출 금지
            break
        try:
            await _drain_one_impl(drain_task, drain_started)
            remaining += 1
        finally:
            queue.task_done()
    return remaining


def _task_order_keys_v2(task: dict) -> frozenset:
    """[2026-10-09] 쓰기 순서 보장 키 — 같은 종목(stock_code)/주문(order_id)의 쓰기는 순서를 지킨다."""
    keys = set()
    data = task.get("data") if isinstance(task, dict) else None
    if isinstance(data, dict):
        if data.get("stock_code"):
            keys.add(("s", data["stock_code"]))
        if data.get("order_id"):
            keys.add(("o", data["order_id"]))
    return frozenset(keys)


class _OrderGateV2:
    """지연 재투입 중인 태스크와 같은 키의 뒤 태스크를 보류해 쓰기 순서를 유지한다.

    재시도가 큐 꼬리로 재투입되면 그 사이 뒤 태스크(close 등)가 먼저 실행돼, 실패했던 진입
    upsert 가 나중에 '행 없음 → 신규 생성' 으로 이미 청산된 포지션의 유령 active 행을 만든다.
    """

    def __init__(self) -> None:
        self._deferred: dict[int, frozenset] = {}   # id(task) -> keys (재시도 대기)
        self._released: dict[int, frozenset] = {}   # id(task) -> keys (풀려났으나 아직 미완료)
        self._held: list[dict] = []

    def defer(self, task: dict) -> None:
        task["_requeued"] = True
        self._released.pop(id(task), None)
        self._deferred[id(task)] = _task_order_keys_v2(task)

    def should_hold(self, task: dict) -> bool:
        tid = id(task)
        if tid in self._deferred:
            return False
        keys = _task_order_keys_v2(task)
        if not keys:
            return False
        if tid in self._released:
            # 풀려난 태스크: 앞선 같은 키 태스크가 다시 재시도 대기에 들어갔다면 앞지르지 않도록 보류
            blocked = any(keys & k for k in self._deferred.values())
            if blocked:
                self._released.pop(tid, None)
            return blocked
        # 재시도 대기 중이거나 풀려났으나 아직 끝나지 않은 같은 키 태스크가 있으면 그 뒤로 보류
        return (any(keys & k for k in self._deferred.values())
                or any(keys & k for k in self._released.values()))

    def hold(self, task: dict) -> None:
        self._held.append(task)

    def held_tasks(self) -> list:
        return list(self._held)

    def resolve(self, task: dict) -> list:
        """태스크가 성공/DLQ 로 끝났을 때 호출. 이제 막힌 곳이 없는 보류 태스크를 순서대로 반환.

        반환된 태스크는 실행이 끝날 때까지 '진행 중' 키로 남아, 큐 꼬리에 뒤따라 들어온
        같은 키의 후속 태스크가 앞지르지 못한다.
        """
        a = self._deferred.pop(id(task), None)
        b = self._released.pop(id(task), None)
        if a is None and b is None:
            return []
        released, still = [], []
        for h in self._held:
            hk = _task_order_keys_v2(h)
            blocked = (any(hk & k for k in self._deferred.values())
                       or any(hk & k for k in self._released.values()))
            if blocked:
                still.append(h)
            else:
                released.append(h)
                self._released[id(h)] = hk
        self._held = still
        return released


async def _requeue_after_v2(queue: asyncio.Queue, task: dict, delay: float, backoff: list) -> None:
    """워커를 막지 않는 지연 재투입. 취소되면 task 는 backoff 목록에 남아 종료 drain 이 처리한다."""
    await asyncio.sleep(delay)
    # remove + put_nowait 사이에 await 가 없어 drain 과 경합하지 않는다
    if task in backoff:
        backoff.remove(task)
    queue.put_nowait(task)
    # [R5] 실패 시 task_done() 을 미뤄 unfinished 를 유지했다 — 재투입분이 +1 이므로 원래 몫을 여기서 상쇄
    queue.task_done()


def _release_held_v2(queue: asyncio.Queue, task: dict) -> None:
    """보류분 재투입. 보류 시 task_done() 을 미뤘으므로 재투입 +1 을 상쇄해 join() 이 보류 중 풀리지 않게 한다."""
    queue.put_nowait(task)
    queue.task_done()


async def _dlq_runtime_replay_loop_v2(queue: asyncio.Queue) -> None:
    """[2026-10-09] 런타임 DLQ 주기 재처리. 재기동 전까지 DLQ 가 방치되던 것을 막는다.

    DB 가 살아 있음을 SELECT 1 로 확인하고, 회로가 열려 있거나 큐가 많이 적체되면 건너뛴다.
    재큐잉 자체는 replay_dlq (큐 경유, 충돌·나이 판정 포함)에 맡긴다.
    """
    from sqlalchemy import text
    while True:
        try:
            await asyncio.wait_for(asyncio.Event().wait(), timeout=DLQ_RUNTIME_REPLAY_INTERVAL_SEC)
        except asyncio.TimeoutError:
            pass
        try:
            if _worker_state["circuit_open"] or queue.qsize() > WORKER_QUEUE_ALERT_DEPTH:
                continue
            counts = await dlq_status_counts()
            if counts.get("pending", 0) <= 0:
                continue
            # DB-READ-EXCEPTION: 헬스 프로브 SELECT 1 (db_worker 소유, 쓰기 없음)
            async with AsyncSessionLocal() as _s:
                await _s.execute(text("SELECT 1"))
            n = await replay_dlq()
            logger.warning("[DLQ] 런타임 재처리: pending=%d → 재큐잉 %d건", counts["pending"], n)
        except asyncio.CancelledError:
            raise
        except Exception as e:
            logger.warning("[DLQ] 런타임 재처리 건너뜀(DB 미복구 또는 오류): %s", e)


async def db_writer_worker() -> None:
    """lifespan 에서 asyncio.create_task() 로 실행되는 영구 백그라운드 워커.

    큐에서 task 를 꺼내 단일 AsyncSession 안에서 처리한 뒤
    commit / rollback 을 자동 관리한다.

    [H-5 2026-10-02] cancel 처리를 루프 **전체**로 넓혔다. 이전에는 CancelledError 를
    태스크 처리 try 의 형제 except 로만 잡아서
      (a) 재시도 백오프 sleep 중 cancel → drain 을 건너뛰고 큐 전체 유실,
      (b) 처리 중이던 태스크는 drain/DLQ 어디에도 남지 않음,
      (c) queue.get() 대기 중 cancel → drain 없이 종료
    가 발생했다.
    """
    logger.info("🛠️  Background DB Writer Worker started")
    queue = get_db_task_queue()  # 워커 진입 시 큐 인스턴스 고정
    setattr(queue, _QUEUE_CLOSED_ATTR, False)
    inflight: dict | None = None
    backoff: list[dict] = []
    pending_saves: set = set()
    timers: set = set()
    circuit_since: float | None = None
    _worker_state.update(consecutive_failures=0, circuit_open=False)
    gate = _OrderGateV2()
    replay_task = asyncio.ensure_future(_dlq_runtime_replay_loop_v2(queue))
    try:
        while True:
            task = await queue.get()
            if gate.should_hold(task):
                # 같은 종목/주문의 앞선 쓰기가 재시도 대기 중 — 순서 보장을 위해 그 뒤로 보류
                # task_done() 을 미룬다(unfinished 유지) — 재투입 시 _release_held_v2 가 상쇄
                gate.hold(task)
                continue
            inflight = task
            _deferred_retry = False
            try:
                _t0 = time.monotonic()
                await _execute_db_task_impl(task)
                inflight = None
                for _rel in gate.resolve(task):
                    _release_held_v2(queue, _rel)
                _worker_state.update(
                    consecutive_failures=0, circuit_open=False, last_success_ts=time.time(),
                )
                circuit_since = None
                _check_queue_backlog_v2(queue, time.monotonic() - _t0)
            except asyncio.CancelledError:
                raise
            except Exception as exc:
                inflight = None
                retry_count = task.get("_retry_count", 0)
                max_retries = 3
                task_type = task.get("type", "unknown")

                _strip_retry_markers(task)  # [M-A] 실패 실행의 절대값 마커는 재사용 금지
                _worker_state["consecutive_failures"] += 1
                if _worker_state["consecutive_failures"] >= CIRCUIT_OPEN_FAILURES:
                    if circuit_since is None:
                        circuit_since = time.monotonic()
                        logger.error(
                            "❌ [Worker] 연속 %d회 실패 — DB 장애로 간주(회로 열림), 재시도 횟수 소모 보류",
                            _worker_state["consecutive_failures"],
                        )
                    _worker_state["circuit_open"] = True
                _circuit_hold = (
                    circuit_since is not None
                    and time.monotonic() - circuit_since < CIRCUIT_MAX_HOLD_SEC
                )
                if queue.qsize() >= WORKER_QUEUE_ALERT_DEPTH:
                    _alert_throttled_v2(
                        "db_worker:queue_depth", "DB_WORKER",
                        f"DB 쓰기 큐 적체 qsize={queue.qsize()} (연속 실패 "
                        f"{_worker_state['consecutive_failures']}회). DB 상태 확인 필요.",
                    )
                if _circuit_hold:
                    # 장애 구간: 재시도 횟수를 소모하지 않고 일괄 대기 후 재투입(DLQ 조기 유실 방지)
                    delay = CIRCUIT_RETRY_DELAY_SEC
                    logger.warning(
                        "⚠️ [Worker] 회로 열림 — type=%s 재시도 횟수 유지(%d), %.0fs 후 재투입: %s",
                        task_type, retry_count, delay, exc,
                    )
                    gate.defer(task)
                    _deferred_retry = True
                    _t = asyncio.ensure_future(_requeue_after_v2(queue, task, delay, backoff))
                    backoff.append(task)
                    timers.add(_t)
                    _t.add_done_callback(timers.discard)
                elif retry_count < max_retries:
                    task["_retry_count"] = retry_count + 1
                    delay = min(2 ** retry_count, 8)  # 1, 2, 4초 지수 백오프
                    logger.warning(
                        "⚠️ [Worker] DB task retry %d/%d (type=%s): %s — %ds 후 재시도",
                        retry_count + 1, max_retries, task_type, exc, delay,
                    )
                    # 워커를 막지 않고 지연 재투입 — 그 사이 다른 태스크 계속 처리.
                    # backoff 목록: 취소 시 drain 이 이 목록을 먼저 처리
                    backoff.append(task)
                    gate.defer(task)
                    _deferred_retry = True
                    _t = asyncio.ensure_future(_requeue_after_v2(queue, task, delay, backoff))
                    timers.add(_t)
                    _t.add_done_callback(timers.discard)
                else:
                    logger.error(
                        "❌ [Worker] DB task DROPPED after %d retries (type=%s): %s",
                        max_retries, task_type, exc, exc_info=True,
                    )
                    # DLQ: 최종 실패 task를 파일로 백업하여 데이터 손실 방지.
                    # shield — 저장 도중 cancel 돼도 파일 쓰기는 끝까지 가고 drain 이 완료를 기다린다.
                    fut = asyncio.ensure_future(_save_to_dlq(task, str(exc)))
                    pending_saves.add(fut)
                    fut.add_done_callback(pending_saves.discard)
                    await asyncio.shield(fut)
                    for _rel in gate.resolve(task):
                        _release_held_v2(queue, _rel)
            finally:
                if not _deferred_retry:
                    queue.task_done()
    except asyncio.CancelledError:
        # 큐 drain: 종료 전 남은 태스크 처리하여 데이터 손실 방지
        for _t in list(timers):
            _t.cancel()
        replay_task.cancel()
        _drain_started = time.monotonic()
        remaining = await _drain_on_shutdown_impl(
            queue, inflight, backoff + [h for h in gate.held_tasks() if h not in backoff],
            pending_saves, drain_started=_drain_started,
        )
        # [M-D] 이후 투입은 DLQ 직행. drain 과 표식 사이에 들어온 건은 아래에서 한 번 더 비운다.
        setattr(queue, _QUEUE_CLOSED_ATTR, True)
        if not queue.empty():
            remaining += await _drain_on_shutdown_impl(
                queue, None, [], set(), drain_started=_drain_started
            )
        if remaining:
            logger.info(f"[DBWorker] 종료 시 {remaining}건 drain 완료")
        logger.info("⏹️  DB Writer Worker stopped")
    finally:
        replay_task.cancel()