예제 소스

text_client_core.py

"""텍스트 VLM 클라이언트 코어 — Qt 거부 정책의 가벼운 미러 + 루프 탈출.

목표:
  - VLM vs VLM 을 headless 로 돌리며 **무한 재요청을 금지**
  - 동일 거부/고착이 나면 대국을 ERROR 로 끝내고 파이프라인이 원인을 모음
  - 규칙 엔진은 `rl_env.board.BoardState` (서버 MiniGo 와 동일 수준, 슈퍼코 없음)
"""

from __future__ import annotations

import hashlib
from dataclasses import dataclass, field
from enum import Enum
from typing import List, Optional, Set, Tuple

from rl_env.board import BLACK, BOARD_SIZE, EMPTY, WHITE, BoardState, DR, DC

Point = Tuple[int, int]


class RejectReason(str, Enum):
    OK = "ok"
    ILLEGAL = "illegal"
    OCCUPIED = "occupied"
    SUICIDE = "suicide"
    LOW_LINE = "low_line"
    SELF_FILL = "self_fill"
    OUT_OF_BOUNDS = "out_of_bounds"


class GameEndReason(str, Enum):
    DOUBLE_PASS = "double_pass"
    MAX_MOVES = "max_moves"
    NO_LEGAL = "no_legal"
    RETRY_EXHAUSTED = "retry_exhausted"
    REJECT_LOOP = "reject_loop"
    STICKY_OFFER = "sticky_offer"
    COMM_ERROR = "comm_error"
    ABORT = "abort"


@dataclass
class SoftDecision:
    accept: bool
    reason: RejectReason
    detail: str = ""


@dataclass
class CoreConfig:
    min_line: int = 3
    max_reject_retries: int = 8
    max_same_reject: int = 3  # 같은 (수, reason) 반복 → 루프
    max_sticky_offers: int = 6  # 보드 불변 + 동일 제안 반복
    max_forbid: int = 48  # GO19 trailer 상한 64에 맞춤 — 저선 고착 완화
    allow_low_line_if_capture: bool = True
    # 통신 일시 실패 시 대국 종료 전 재연결 횟수 (타임아웃·Connection reset 등)
    max_comm_retries: int = 3
    comm_retry_sleep_s: float = 2.0


@dataclass
class ClientCore:
    """한 대국 상태. 착수 수용/거부/루프 감지만 담당 (네트워크 없음)."""

    cfg: CoreConfig = field(default_factory=CoreConfig)
    board: BoardState = field(default_factory=BoardState)
    to_play: int = BLACK
    ply: int = 0
    pass_streak: int = 0
    game_epoch: int = 1
    forbid: List[Point] = field(default_factory=list)
    # 루프 감지
    _reject_counts: dict = field(default_factory=dict)  # (r,c,reason) -> n
    _sticky: dict = field(default_factory=dict)  # (side, r, c, board_fp) -> n
    _turn_retries: int = 0
    last_error: Optional[str] = None

    def board_fp(self) -> str:
        return hashlib.sha1(self.board.packed()).hexdigest()[:12]

    def edge_line(self, r: int, c: int) -> int:
        N = self.board.size
        return min(r, c, N - 1 - r, N - 1 - c) + 1

    def _would_capture(self, r: int, c: int, who: int) -> bool:
        N = self.board.size
        opp = WHITE if who == BLACK else BLACK
        if self.board.at(r, c) != EMPTY:
            return False
        snapshot = bytes(self.board.cells)
        self.board._set(r, c, who)
        captured = False
        seen = bytearray(N * N)
        for d in range(4):
            nr, nc = r + DR[d], c + DC[d]
            if not (0 <= nr < N and 0 <= nc < N):
                continue
            nidx = nr * N + nc
            if self.board.cells[nidx] != opp or seen[nidx]:
                continue
            stones, libs = self.board._group(nr, nc)
            for s in stones:
                seen[s] = 1
            if libs == 0:
                captured = True
                break
        self.board.cells[:] = snapshot
        return captured

    def _is_self_fill(self, r: int, c: int, who: int) -> bool:
        """Qt selfFill 근사: 내 돌 ≥3 · 상대 0 · 따냄/단수탈출 아님."""
        N = self.board.size
        if self.board.at(r, c) != EMPTY:
            return False
        if self._would_capture(r, c, who):
            return False
        opp = WHITE if who == BLACK else BLACK
        own = opp_n = 0
        rescues = False
        for d in range(4):
            nr, nc = r + DR[d], c + DC[d]
            if not (0 <= nr < N and 0 <= nc < N):
                continue
            v = self.board.at(nr, nc)
            if v == who:
                own += 1
                _, libs = self.board._group(nr, nc)
                if libs == 1:
                    rescues = True
            elif v == opp:
                opp_n += 1
        return own >= 3 and opp_n == 0 and not rescues

    def evaluate_move(self, r: Optional[int], c: Optional[int], *, is_pass: bool) -> SoftDecision:
        who = self.to_play
        if is_pass or r is None or c is None:
            return SoftDecision(True, RejectReason.OK, "pass")
        ill = self.board.illegal_reason(r, c, who)
        if ill is not None:
            mapping = {
                "occupied": RejectReason.OCCUPIED,
                "suicide": RejectReason.SUICIDE,
                "out_of_bounds": RejectReason.OUT_OF_BOUNDS,
            }
            return SoftDecision(False, mapping.get(ill, RejectReason.ILLEGAL), ill)
        cap = self._would_capture(r, c, who)
        if self.edge_line(r, c) < self.cfg.min_line and not (
            self.cfg.allow_low_line_if_capture and cap
        ):
            return SoftDecision(False, RejectReason.LOW_LINE, f"L{self.edge_line(r, c)}")
        if self._is_self_fill(r, c, who):
            return SoftDecision(False, RejectReason.SELF_FILL, "self_fill")
        return SoftDecision(True, RejectReason.OK, "")

    def note_forbid(self, r: int, c: int) -> None:
        if (r, c) in self.forbid:
            return
        self.forbid.append((r, c))
        if len(self.forbid) > self.cfg.max_forbid:
            self.forbid = self.forbid[-self.cfg.max_forbid :]

    def clear_turn_state(self) -> None:
        self._turn_retries = 0
        # forbid 는 턴 성공 시 비움
        self.forbid.clear()
        self._reject_counts.clear()

    def on_reject(self, r: int, c: int, reason: RejectReason) -> Optional[GameEndReason]:
        """거부 기록. 루프면 GameEndReason 반환."""
        self.note_forbid(r, c)
        self._turn_retries += 1
        key = (r, c, reason.value)
        self._reject_counts[key] = self._reject_counts.get(key, 0) + 1
        if self._reject_counts[key] >= self.cfg.max_same_reject:
            self.last_error = f"reject_loop {reason.value} at ({r},{c}) x{self._reject_counts[key]}"
            return GameEndReason.REJECT_LOOP
        if self._turn_retries >= self.cfg.max_reject_retries:
            self.last_error = f"retry_exhausted after {self._turn_retries} (last={reason.value} {r},{c})"
            return GameEndReason.RETRY_EXHAUSTED
        return None

    def on_server_offer(self, r: int, c: int) -> Optional[GameEndReason]:
        """서버가 같은 보드에서 같은 수를 반복 제안하면 STICKY."""
        fp = self.board_fp()
        key = (self.to_play, r, c, fp)
        self._sticky[key] = self._sticky.get(key, 0) + 1
        if self._sticky[key] >= self.cfg.max_sticky_offers:
            self.last_error = f"sticky_offer side={self.to_play} ({r},{c}) fp={fp}"
            return GameEndReason.STICKY_OFFER
        return None

    def apply_pass(self) -> Optional[GameEndReason]:
        self.pass_streak += 1
        self.ply += 1
        self.to_play = WHITE if self.to_play == BLACK else BLACK
        self.clear_turn_state()
        if self.pass_streak >= 2:
            return GameEndReason.DOUBLE_PASS
        return None

    def apply_move(self, r: int, c: int) -> None:
        self.board.play(r, c, self.to_play)
        self.pass_streak = 0
        self.ply += 1
        self.to_play = WHITE if self.to_play == BLACK else BLACK
        self.clear_turn_state()
        # sticky 키는 보드가 바뀌면 자연히 무효 (fp 변경)

    def any_legal(self) -> Optional[Point]:
        N = self.board.size
        who = self.to_play
        for r in range(N):
            for c in range(N):
                if self.board.is_legal(r, c, who):
                    # soft 도 OK 인 것만
                    d = self.evaluate_move(r, c, is_pass=False)
                    if d.accept:
                        return (r, c)
        # soft 통과가 없으면 합법만이라도
        for r in range(N):
            for c in range(N):
                if self.board.is_legal(r, c, who):
                    return (r, c)
        return None

    def ascii_board(self) -> str:
        N = self.board.size
        rows = []
        for r in range(N):
            cells = []
            for c in range(N):
                v = self.board.at(r, c)
                cells.append("." if v == EMPTY else ("X" if v == BLACK else "O"))
            rows.append(f"{r:2d} " + " ".join(cells))
        return "\n".join(rows)