예제 소스
SGF 기보로 학습하는 바둑 LLM 인공지능 · 파일 169개
monitor_mariago_web_play.py
#!/usr/bin/env python3
"""MariaGO 웹 경로 모니터링 대국 (텍스트).
웹과 동일하게 WebSocket 브리지(:9320) → VLM(:9319) 경로로 둡니다.
흑=휴리스틱, 백=VLM. 불법수·타임아웃·패스 루프·계가 이상을 기록합니다.
.venv/bin/python scripts/monitor_mariago_web_play.py --games 3 --max-moves 200
"""
from __future__ import annotations
import argparse
import asyncio
import json
import random
import sys
import time
from collections import Counter
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import List, Optional, Tuple
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from vlm_server.cross_lessons import ( # noqa: E402
LESSON_PATH,
absorb_game,
ingest_jsonl,
lesson_bonus_heuristic,
load_book,
r5_live_delta,
save_book,
)
from scripts.match_db import ( # noqa: E402
FinalJudge,
MatchDB,
ServerLogTail,
board_hash,
default_server_log,
)
BOARD_SIZE = 19
EMPTY, BLACK, WHITE = 0, 1, 2
HEURISTIC_VERSION = "cross-v2"
MAX_GAME_PLIES = 200
# --max-moves 0(무제한)에서도 단순 패 반복 등으로 끝나지 않는 판을 막는 상한.
SAFETY_PLIES = 1000
# 남은 수가 모두 자충·호구·헛도망 수준의 손해수면 휴리스틱도 PASS한다.
HEURISTIC_PASS_FLOOR = -4000
# 이 크기 이하의 빈 영역이 내 돌에만 닿아 있으면 내 집으로 보고 메우지 않는다.
OWN_REGION_MAX = 40
MIDGAME_EMPTY_MIN = 120
DR = (-1, 1, 0, 0)
DC = (0, 0, -1, 1)
Point = Tuple[int, int]
def _iso() -> str:
return datetime.now(timezone.utc).isoformat()
def idx(r: int, c: int) -> int:
return r * BOARD_SIZE + c
def in_bounds(r: int, c: int) -> bool:
return 0 <= r < BOARD_SIZE and 0 <= c < BOARD_SIZE
def empty_board() -> List[int]:
return [EMPTY] * (BOARD_SIZE * BOARD_SIZE)
def group_and_libs(board: List[int], r: int, c: int) -> Tuple[List[int], set]:
color = board[idx(r, c)]
stones: List[int] = []
libs: set = set()
seen = set()
stack = [idx(r, c)]
seen.add(idx(r, c))
while stack:
cur = stack.pop()
stones.append(cur)
rr, cc = divmod(cur, BOARD_SIZE)
for d in range(4):
nr, nc = rr + DR[d], cc + DC[d]
if not in_bounds(nr, nc):
continue
n = idx(nr, nc)
v = board[n]
if v == EMPTY:
libs.add(n)
elif v == color and n not in seen:
seen.add(n)
stack.append(n)
return stones, libs
def try_play(
board: List[int], r: int, c: int, who: int, ko: Optional[Point]
) -> Tuple[bool, str, List[int], int, Optional[Point]]:
if not in_bounds(r, c):
return False, "out_of_bounds", board, 0, ko
if board[idx(r, c)] != EMPTY:
return False, "occupied", board, 0, ko
if ko is not None and ko == (r, c):
return False, "ko", board, 0, ko
nxt = board[:]
nxt[idx(r, c)] = who
opp = WHITE if who == BLACK else BLACK
captured = 0
captured_pts: List[int] = []
visited = set()
for d in range(4):
nr, nc = r + DR[d], c + DC[d]
if not in_bounds(nr, nc):
continue
n = idx(nr, nc)
if nxt[n] != opp or n in visited:
continue
stones, libs = group_and_libs(nxt, nr, nc)
visited.update(stones)
if not libs:
for s in stones:
nxt[s] = EMPTY
captured_pts.append(s)
captured += 1
stones, libs = group_and_libs(nxt, r, c)
if not libs:
return False, "suicide", board, 0, ko
new_ko = None
if captured == 1 and len(stones) == 1:
p = captured_pts[0]
new_ko = divmod(p, BOARD_SIZE)
return True, "ok", nxt, captured, new_ko
def score_korean(board: List[int], cap_b: int, cap_w: int, komi: float = 6.5) -> dict:
seen = [0] * (BOARD_SIZE * BOARD_SIZE)
terr_b = terr_w = dame = 0
for i in range(len(board)):
if board[i] != EMPTY or seen[i]:
continue
stack = [i]
seen[i] = 1
region = []
touch = set()
while stack:
cur = stack.pop()
region.append(cur)
rr, cc = divmod(cur, BOARD_SIZE)
for d in range(4):
nr, nc = rr + DR[d], cc + DC[d]
if not in_bounds(nr, nc):
continue
n = idx(nr, nc)
v = board[n]
if v == EMPTY and not seen[n]:
seen[n] = 1
stack.append(n)
elif v in (BLACK, WHITE):
touch.add(v)
if touch == {BLACK}:
terr_b += len(region)
elif touch == {WHITE}:
terr_w += len(region)
else:
dame += len(region)
b_score = terr_b + cap_b
w_score = terr_w + cap_w + komi
if b_score > w_score:
winner = "black"
elif w_score > b_score:
winner = "white"
else:
winner = "draw"
return {
"black_territory": terr_b,
"white_territory": terr_w,
"black_captures": cap_b,
"white_captures": cap_w,
"komi": komi,
"black_score": b_score,
"white_score": w_score,
"winner": winner,
"dame": dame,
}
def edge_line(r: int, c: int) -> int:
return min(r, c, BOARD_SIZE - 1 - r, BOARD_SIZE - 1 - c) + 1
def _adjacent_groups(board: List[int], r: int, c: int, color: int) -> List[Tuple[List[int], set]]:
groups: List[Tuple[List[int], set]] = []
seen = set()
for d in range(4):
nr, nc = r + DR[d], c + DC[d]
if not in_bounds(nr, nc):
continue
n = idx(nr, nc)
if board[n] != color or n in seen:
continue
stones, libs = group_and_libs(board, nr, nc)
seen.update(stones)
groups.append((stones, libs))
return groups
def _opening_shape_score(board: List[int], r: int, c: int) -> int:
"""초반에는 귀의 3·4선을 우선하고 의미 없는 중앙 산개를 억제한다."""
move_count = sum(1 for v in board if v != EMPTY)
if move_count >= 70:
return 0
line = edge_line(r, c)
if line == 4:
score = 900
elif line == 3:
score = 700
elif line == 2:
score = -700
elif line == 1:
score = -1800
else:
score = 0
star_points = ((3, 3), (3, 15), (15, 3), (15, 15))
star_distance = min(abs(r - sr) + abs(c - sc) for sr, sc in star_points)
score += max(0, 600 - star_distance * 120)
occupied = [divmod(i, BOARD_SIZE) for i, v in enumerate(board) if v != EMPTY]
if occupied:
nearest = min(abs(r - rr) + abs(c - cc) for rr, cc in occupied)
if nearest == 1:
score -= 900
elif nearest == 2:
score -= 250
elif 3 <= nearest <= 7:
score += 220
return score
def _framework_score(board: List[int], r: int, c: int, who: int) -> int:
"""고립된 산개보다 기존 돌과 느슨하게 연결된 세력 확장을 우선한다."""
own = [divmod(i, BOARD_SIZE) for i, v in enumerate(board) if v == who]
if len(own) < 4:
return 0
nearest = min(abs(r - rr) + abs(c - cc) for rr, cc in own)
score = 0
if len(own) >= 6:
if nearest == 1:
score += 80
elif nearest <= 3:
score += 300
elif nearest <= 5:
score += 140
elif nearest >= 8:
score -= 300
relation = 0
for rr, cc in own:
dr, dc = abs(r - rr), abs(c - cc)
if (dr, dc) in ((0, 2), (2, 0)):
relation = max(relation, 320) # 한 칸 뜀
elif (dr, dc) in ((1, 2), (2, 1)):
relation = max(relation, 270) # 날일자
elif (dr, dc) in ((0, 3), (3, 0)):
relation = max(relation, 180) # 두 칸 벌림
score += relation
left = any(rr == r and 0 < c - cc <= 5 for rr, cc in own)
right = any(rr == r and 0 < cc - c <= 5 for rr, cc in own)
up = any(cc == c and 0 < r - rr <= 5 for rr, cc in own)
down = any(cc == c and 0 < rr - r <= 5 for rr, cc in own)
if (left and right) or (up and down):
score += 220
return score
def _opponent_can_force_capture(
board: List[int],
protected_stones: List[int],
who: int,
ko: Optional[Point],
) -> bool:
"""활로 둘인 구조가 상대 단수 뒤 확실히 잡히는지 한 수 더 확인한다."""
if not protected_stones:
return False
anchor = next((s for s in protected_stones if board[s] == who), None)
if anchor is None:
return True
_stones, libs = group_and_libs(board, *divmod(anchor, BOARD_SIZE))
if len(libs) != 2:
return False
opp = WHITE if who == BLACK else BLACK
protected = set(protected_stones)
for attack in libs:
ar, ac = divmod(attack, BOARD_SIZE)
ok, _reason, attacked, _captured, next_ko = try_play(board, ar, ac, opp, ko)
if not ok:
continue
survivors = [s for s in protected if attacked[s] == who]
if not survivors:
return True
_group, remaining = group_and_libs(attacked, *divmod(survivors[0], BOARD_SIZE))
if len(remaining) != 1:
continue
can_escape = False
for rr in range(BOARD_SIZE):
for cc in range(BOARD_SIZE):
legal, _why, escaped, _cap, _ko = try_play(attacked, rr, cc, who, next_ko)
if not legal:
continue
survivor = next((s for s in protected if escaped[s] == who), None)
if survivor is None:
continue
_saved, saved_libs = group_and_libs(escaped, *divmod(survivor, BOARD_SIZE))
if len(saved_libs) >= 2:
can_escape = True
break
if can_escape:
break
if not can_escape:
return True
return False
def _heuristic_score(
board: List[int],
r: int,
c: int,
who: int,
ko: Optional[Point],
) -> Optional[int]:
ok, _reason, nxt, captured, new_ko = try_play(board, r, c, who, ko)
if not ok:
return None
opp = WHITE if who == BLACK else BLACK
own_before = _adjacent_groups(board, r, c, who)
rescued_atari = any(idx(r, c) in libs and len(libs) == 1 for _stones, libs in own_before)
score = captured * 10_000
# 단수인 우리 돌의 유일한 활로에 두는 수는 따냄 다음으로 긴급하다.
for stones, libs in own_before:
if idx(r, c) not in libs:
continue
if len(libs) == 1:
score += 8_000 + len(stones) * 700
elif len(libs) == 2:
score += 700 + len(stones) * 80
if len(own_before) >= 2:
score += (len(own_before) - 1) * 500
own_after, own_libs = group_and_libs(nxt, r, c)
if rescued_atari and _opponent_can_force_capture(nxt, own_after, who, new_ko):
score -= 12_000 + len(own_after) * 500
if captured == 0 and len(own_libs) == 1:
score -= 9_000 + len(own_after) * 500
elif captured == 0 and len(own_libs) == 2:
score -= 500 + len(own_after) * 30
else:
score += min(len(own_libs), 6) * 35
# 상대 약한 돌을 단수로 몰거나 활로를 줄이는 수를 우선한다.
for stones, libs in _adjacent_groups(nxt, r, c, opp):
if len(libs) == 1:
score += 1_400 + len(stones) * 180
elif len(libs) == 2:
score += 480 + len(stones) * 50
own_neighbors = sum(
1
for d in range(4)
if in_bounds(r + DR[d], c + DC[d]) and board[idx(r + DR[d], c + DC[d])] == who
)
opp_neighbors = sum(
1
for d in range(4)
if in_bounds(r + DR[d], c + DC[d]) and board[idx(r + DR[d], c + DC[d])] == opp
)
score += own_neighbors * 120 + opp_neighbors * 100
score += _opening_shape_score(board, r, c)
score += _framework_score(board, r, c, who)
score += r5_live_delta(board, r, c, who)
score += lesson_bonus_heuristic(board, r, c, who)
return score
def _rescuable_atari_groups(
board: List[int],
who: int,
ko: Optional[Point],
) -> List[List[int]]:
groups: List[List[int]] = []
seen = set()
for i, value in enumerate(board):
if value != who or i in seen:
continue
stones, libs = group_and_libs(board, *divmod(i, BOARD_SIZE))
seen.update(stones)
if len(libs) != 1:
continue
liberty = next(iter(libs))
rr, cc = divmod(liberty, BOARD_SIZE)
ok, _reason, escaped, _captured, next_ko = try_play(board, rr, cc, who, ko)
if not ok:
continue
survivor = next((s for s in stones if escaped[s] == who), None)
if survivor is None:
continue
saved, saved_libs = group_and_libs(escaped, *divmod(survivor, BOARD_SIZE))
if len(saved_libs) < 2:
continue
if _opponent_can_force_capture(escaped, saved, who, next_ko):
continue
groups.append(stones)
return groups
def _own_region_point(board: List[int], r: int, c: int, who: int) -> bool:
"""(r,c)가 속한 빈 영역이 작고 내 돌에만 닿으면 True (내 집 안쪽)."""
start = idx(r, c)
if board[start] != EMPTY:
return False
seen = {start}
stack = [start]
while stack:
cur = stack.pop()
rr, cc = divmod(cur, BOARD_SIZE)
for d in range(4):
nr, nc = rr + DR[d], cc + DC[d]
if not in_bounds(nr, nc):
continue
n = idx(nr, nc)
v = board[n]
if v == EMPTY:
if n not in seen:
if len(seen) >= OWN_REGION_MAX:
return False
seen.add(n)
stack.append(n)
elif v != who:
return False
return True
def _fills_own_region(board: List[int], r: int, c: int, who: int, ko: Optional[Point]) -> bool:
"""따냄·단수 구원이 아닌데 내 집을 메우는 수."""
if not _own_region_point(board, r, c, who):
return False
ok, _reason, _nxt, captured, _ko = try_play(board, r, c, who, ko)
if not ok or captured > 0:
return False
for _stones, libs in _adjacent_groups(board, r, c, who):
if len(libs) == 1 and idx(r, c) in libs:
return False
return True
def heuristic_move(
board: List[int],
who: int,
ko: Optional[Point],
rng: random.Random,
banned: Optional[set] = None,
) -> Tuple[Optional[int], Optional[int], bool]:
"""전술 안전성·연결·귀 포석을 반영한 경량 휴리스틱.
내 집 메우기는 후보에서 빼고, 남은 수가 모두 큰 손해수면 PASS해서
수 제한 없이도 종국(쌍패스)에 도달하게 한다.
"""
urgent_groups = _rescuable_atari_groups(board, who, ko)
candidates = []
for r in range(BOARD_SIZE):
for c in range(BOARD_SIZE):
if banned and (r, c) in banned:
continue
if _fills_own_region(board, r, c, who, ko):
continue
score = _heuristic_score(board, r, c, who, ko)
if score is None:
continue
if urgent_groups:
_ok, _reason, nxt, _captured, _new_ko = try_play(board, r, c, who, ko)
for stones in urgent_groups:
survivor = next((s for s in stones if nxt[s] == who), None)
safe = False
if survivor is not None:
_saved, libs = group_and_libs(nxt, *divmod(survivor, BOARD_SIZE))
safe = len(libs) >= 2
if not safe:
score -= len(stones) * 2_500
candidates.append((score, r, c))
if not candidates:
return None, None, True
best_score = max(item[0] for item in candidates)
if best_score < HEURISTIC_PASS_FLOOR:
return None, None, True
best = [(r, c) for score, r, c in candidates if score == best_score]
r, c = rng.choice(best)
return r, c, False
def coord(r: Optional[int], c: Optional[int], is_pass: bool = False) -> str:
if is_pass or r is None or c is None:
return "PASS"
letters = "ABCDEFGHJKLMNOPQRST"
return f"{letters[c]}{BOARD_SIZE - r} ({r},{c})"
@dataclass
class Issue:
kind: str
detail: str
ply: int
@dataclass
class GameReport:
game_id: int
end_reason: str
plies: int
issues: List[Issue] = field(default_factory=list)
score: Optional[dict] = None
moves_summary: List[str] = field(default_factory=list)
started_at: str = ""
finished_at: str = ""
pass_checks: List[dict] = field(default_factory=list)
final_board: List[int] = field(default_factory=list)
katago_lead_b: Optional[float] = None
async def ws_request_move(
ws_url: str,
board: List[int],
ai_side: int,
game_epoch: int,
rejects: List[Point],
timeout_s: float,
) -> dict:
try:
import websockets
except ImportError:
return {"ok": False, "error": "websockets not installed"}
try:
async with websockets.connect(ws_url, open_timeout=10, close_timeout=5) as ws:
await ws.send(
json.dumps(
{
"type": "move",
"board": board,
"ai_side": ai_side,
"game_epoch": game_epoch,
"rejects": [list(p) for p in rejects],
}
)
)
raw = await asyncio.wait_for(ws.recv(), timeout=timeout_s)
msg = json.loads(raw)
if msg.get("type") == "error":
return {"ok": False, "error": msg.get("error") or "error"}
if msg.get("type") != "move":
return {"ok": False, "error": f"bad type {msg.get('type')}"}
if msg.get("pass"):
return {"ok": True, "pass": True}
return {"ok": True, "pass": False, "row": int(msg["row"]), "col": int(msg["col"])}
except Exception as e:
return {"ok": False, "error": str(e)}
def resolve_max_moves(max_moves: int) -> int:
"""0 이하는 무제한(SAFETY_PLIES 상한), 양수는 기존처럼 200수까지."""
if int(max_moves) <= 0:
return SAFETY_PLIES
return max(1, min(int(max_moves), MAX_GAME_PLIES))
async def recheck_pass(
ws_url: str,
board: List[int],
ko: Optional[Point],
epoch: int,
rejects: List[Point],
timeout_s: float,
*,
ply: int,
first_dt: float,
) -> Tuple[Optional[Point], dict]:
"""R5 PASS 직후 같은 국면을 한 번 더 요청해 PASS가 재현되는지 본다.
Returns: (두 번째 응답이 합법 착수면 좌표, 기록용 dict)
"""
empty_n = sum(1 for x in board if x == EMPTY)
t0 = time.time()
second = await ws_request_move(ws_url, board, WHITE, epoch, rejects, timeout_s)
dt = time.time() - t0
rec = {
"ply": ply,
"empty": empty_n,
"stones": BOARD_SIZE * BOARD_SIZE - empty_n,
"phase": "midgame" if empty_n >= MIDGAME_EMPTY_MIN else "endgame",
"first_dt": round(first_dt, 2),
"second_dt": round(dt, 2),
"board": list(board),
}
if not second.get("ok"):
rec["result"] = "recheck_error"
rec["error"] = str(second.get("error"))
return None, rec
if second.get("pass"):
rec["result"] = "pass_confirmed"
return None, rec
wr, wc = int(second["row"]), int(second["col"])
ok, reason, *_ = try_play(board, wr, wc, WHITE, ko)
rec["second_move"] = [wr, wc]
if not ok:
rec["result"] = "flipped_illegal"
rec["illegal_reason"] = reason
return None, rec
rec["result"] = "pass_flipped"
return (wr, wc), rec
async def play_one(
game_id: int,
ws_url: str,
max_moves: int,
timeout_s: float,
seed: int,
pass_recheck: bool = False,
db: Optional[MatchDB] = None,
tail: Optional[ServerLogTail] = None,
) -> GameReport:
rng = random.Random(seed + game_id)
board = empty_board()
ko: Optional[Point] = None
to_play = BLACK
cap_b = cap_w = 0
pass_streak = 0
rejects: List[Point] = []
epoch = int(time.time()) + game_id
issues: List[Issue] = []
summary: List[str] = []
rep = GameReport(game_id=game_id, end_reason="max_moves", plies=0, started_at=_iso())
unlimited = int(max_moves) <= 0
max_moves = resolve_max_moves(max_moves)
if db is not None:
db.start_game(game_id, epoch, rep.started_at)
def record(ply: int, color: str, kind: str, before: List[int], **kw) -> None:
if db is None:
return
server_rec = None
if color == "W" and tail is not None:
server_rec = tail.grab(board_hash(before, WHITE))
db.add_move(
game_id=game_id,
ply=ply,
color=color,
kind=kind,
board=before,
ts=_iso(),
server_rec=server_rec,
**kw,
)
for ply in range(max_moves):
if to_play == BLACK:
r, c, is_pass = heuristic_move(board, BLACK, ko, rng)
record(ply, "B", "pass" if is_pass else "move", board, row=r, col=c)
if is_pass:
pass_streak += 1
summary.append(f"{ply+1}. B PASS")
if pass_streak >= 2:
rep.end_reason = "double_pass"
break
to_play = WHITE
continue
ok, reason, nxt, cap, new_ko = try_play(board, r, c, BLACK, ko)
if not ok:
issues.append(Issue("heuristic_illegal", reason, ply))
pass_streak += 1
to_play = WHITE
continue
board, ko = nxt, new_ko
cap_b += cap
pass_streak = 0
rejects = []
summary.append(f"{ply+1}. B {coord(r, c)}")
to_play = WHITE
else:
t0 = time.time()
reply = await ws_request_move(ws_url, board, WHITE, epoch, rejects, timeout_s)
dt = time.time() - t0
if not reply.get("ok"):
record(ply, "W", "error", board, dt=dt)
issues.append(Issue("vlm_comm", str(reply.get("error")), ply))
rep.end_reason = "comm_error"
summary.append(f"{ply+1}. W ERROR {reply.get('error')}")
break
first_kind = "pass"
if not reply.get("pass"):
legal = try_play(board, int(reply["row"]), int(reply["col"]), WHITE, ko)[0]
first_kind = "move" if legal else "illegal"
record(
ply,
"W",
first_kind,
board,
row=reply.get("row"),
col=reply.get("col"),
dt=dt,
)
recheck_tag = ""
if reply.get("pass") and pass_recheck:
flip, rec = await recheck_pass(
ws_url, board, ko, epoch, rejects, timeout_s, ply=ply, first_dt=dt
)
second = rec.get("second_move") or [None, None]
record(
ply,
"W",
"pass" if rec["result"] == "pass_confirmed" else rec["result"],
board,
row=second[0],
col=second[1],
dt=rec.get("second_dt"),
attempt="recheck",
)
rep.pass_checks.append(rec)
issues.append(
Issue(rec["result"], f"{rec['phase']} empty={rec['empty']} ply={ply}", ply)
)
if flip is not None:
reply = {"ok": True, "pass": False, "row": flip[0], "col": flip[1]}
dt += rec["second_dt"]
recheck_tag = ", recheck"
if reply.get("pass"):
empty_n = sum(1 for x in board if x == EMPTY)
if empty_n >= 250 or ply < 40:
issues.append(
Issue(
"early_vlm_pass",
f"초반/빈집 과다 PASS empty={empty_n} ply={ply} dt={dt:.1f}s",
ply,
)
)
pass_streak += 1
summary.append(f"{ply+1}. W PASS ({dt:.1f}s)")
if pass_streak >= 2:
rep.end_reason = "double_pass"
if empty_n >= 250 or ply < 40:
issues.append(
Issue(
"early_double_pass",
f"초반 쌍패스 종료 empty={empty_n} ply={ply}",
ply,
)
)
break
to_play = BLACK
continue
wr, wc = int(reply["row"]), int(reply["col"])
ok, reason, nxt, cap, new_ko = try_play(board, wr, wc, WHITE, ko)
if not ok:
issues.append(Issue("vlm_illegal", f"{reason} at {coord(wr, wc)}", ply))
rejects = (rejects + [(wr, wc)])[-48:]
# 웹과 같이 1회 재요청
retry = await ws_request_move(ws_url, board, WHITE, epoch, rejects, timeout_s)
record(
ply,
"W",
"error" if not retry.get("ok") else ("pass" if retry.get("pass") else "move"),
board,
row=retry.get("row"),
col=retry.get("col"),
attempt="illegal_retry",
)
if not retry.get("ok"):
issues.append(Issue("vlm_retry_comm", str(retry.get("error")), ply))
rep.end_reason = "comm_error"
break
if retry.get("pass"):
empty_n = sum(1 for x in board if x == EMPTY)
issues.append(
Issue(
"vlm_pass_after_illegal",
f"불법수 후 PASS empty={empty_n} ply={ply}",
ply,
)
)
if empty_n >= 250 or ply < 40:
issues.append(
Issue(
"early_vlm_pass",
f"불법수 후 초반 PASS empty={empty_n}",
ply,
)
)
pass_streak += 1
summary.append(f"{ply+1}. W PASS after illegal ({dt:.1f}s)")
if pass_streak >= 2:
rep.end_reason = "double_pass"
break
to_play = BLACK
continue
wr, wc = int(retry["row"]), int(retry["col"])
ok, reason, nxt, cap, new_ko = try_play(board, wr, wc, WHITE, ko)
if not ok:
issues.append(Issue("vlm_illegal_retry", f"{reason} at {coord(wr, wc)}", ply))
rep.end_reason = "vlm_illegal_loop"
summary.append(f"{ply+1}. W ILLEGAL×2 {coord(wr, wc)}")
break
board, ko = nxt, new_ko
cap_w += cap
pass_streak = 0
rejects = []
summary.append(f"{ply+1}. W {coord(wr, wc)} ({dt:.1f}s{recheck_tag})")
to_play = BLACK
# 연속 PASS 과다 경고는 end에서
if len(summary) >= 12:
recent = summary[-12:]
if sum(1 for s in recent if "PASS" in s) >= 10:
issues.append(Issue("pass_storm", "PASS 과다 (최근 12수 중 10+)", ply))
rep.end_reason = "pass_storm"
break
rep.plies = len(summary)
rep.moves_summary = summary
rep.final_board = list(board)
rep.score = score_korean(board, cap_b, cap_w)
if rep.end_reason == "max_moves" and unlimited:
rep.end_reason = "safety_cap"
issues.append(Issue("hit_safety_cap", f"safety_plies={max_moves}", rep.plies))
elif rep.end_reason == "max_moves":
issues.append(Issue("hit_max_moves", f"max_moves={max_moves}", rep.plies))
# 계가 이상: dame가 판의 대부분이면 미정국
if rep.score and rep.score["dame"] > 200 and rep.plies > 40:
issues.append(Issue("score_uncertain", f"dame={rep.score['dame']} (사활 미판정)", rep.plies))
rep.issues = issues
rep.finished_at = _iso()
return rep
def _write_status(
*,
out: Path,
games_path: Path,
issues_path: Path,
live_path: Path,
gid: int,
games_target: int,
rep: GameReport,
issue_counts: Counter,
end_counts: Counter,
winner_counts: Optional[Counter] = None,
recheck_counts: Optional[Counter] = None,
) -> None:
meaningful = [i for i in rep.issues if i.kind != "hit_max_moves"]
tag = "ISSUE" if meaningful else ("ok" if not rep.issues else "cap")
print(
f"[monitor] game {gid} {tag} end={rep.end_reason} plies={rep.plies} "
f"issues={[i.kind for i in meaningful] or [i.kind for i in rep.issues]} "
f"score={rep.score and rep.score.get('winner')}",
flush=True,
)
with games_path.open("a", encoding="utf-8") as f:
f.write(
json.dumps(
{
"ts": _iso(),
"game_id": rep.game_id,
"end_reason": rep.end_reason,
"plies": rep.plies,
"score": rep.score,
"issues": [asdict(i) for i in rep.issues],
"heuristic_version": HEURISTIC_VERSION,
"moves": rep.moves_summary,
"moves_tail": rep.moves_summary[-20:],
"pass_checks": [
{k: v for k, v in pc.items() if k != "board"} for pc in rep.pass_checks
],
},
ensure_ascii=False,
)
+ "\n"
)
target = "∞" if games_target <= 0 else str(games_target)
live_path.write_text(
"\n".join(
[
f"# MariaGO web monitor — {_iso()}",
"",
f"- game: `{gid}/{target}`",
f"- last_end: `{rep.end_reason}`",
f"- issue_counts: `{dict(issue_counts)}`",
f"- end_counts: `{dict(end_counts)}`",
"",
"## last moves",
"",
*([f"- {m}" for m in rep.moves_summary[-15:]] or ["_none_"]),
"",
]
),
encoding="utf-8",
)
lines = [
f"# MariaGO web monitor ISSUES — {_iso()}",
"",
f"경로: 휴리스틱(흑) → WS (웹과 동일) — 조기 PASS·불법수 추적",
"",
f"- games_done: `{gid}`",
"",
"## end_reason",
]
for k, v in end_counts.most_common():
lines.append(f"- `{k}`: {v}")
if winner_counts:
lines += ["", "## winner (흑=휴리스틱, 백=R5)"]
for k, v in winner_counts.most_common():
lines.append(f"- `{k}`: {v}")
if recheck_counts:
lines += ["", "## R5 PASS 재요청 결과 (phase:result)"]
for k, v in sorted(recheck_counts.items()):
lines.append(f"- `{k}`: {v}")
lines += ["", "## issue kinds"]
if issue_counts:
for k, v in issue_counts.most_common():
lines.append(f"- `{k}`: {v}")
else:
lines.append("- (없음)")
lines += ["", f"상세: `{games_path.name}`", f"라이브: `{live_path.name}`"]
issues_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
async def amain(args: argparse.Namespace) -> int:
out = Path(args.out_dir)
if not out.is_absolute():
out = ROOT / out
out.mkdir(parents=True, exist_ok=True)
games_path = out / "games.jsonl"
issues_path = out / "ISSUES.md"
live_path = out / "LIVE.md"
heart_path = out / "HEARTBEAT.md"
pass_cases_path = out / "pass_cases.jsonl"
issue_counts: Counter = Counter()
end_counts: Counter = Counter()
winner_counts: Counter = Counter()
recheck_counts: Counter = Counter()
infinite = args.games <= 0
max_label = "∞" if int(args.max_moves) <= 0 else str(resolve_max_moves(args.max_moves))
lesson_book = ingest_jsonl(games_path, load_book(LESSON_PATH), source=out.name)
prior = ROOT / "logs" / "heuristic_vs_r5_150_20260928" / "games.jsonl"
if prior.resolve() != games_path.resolve():
lesson_book = ingest_jsonl(prior, lesson_book, source=prior.parent.name)
save_book(lesson_book, LESSON_PATH)
print(
f"[monitor] ws={args.ws_url} games={'∞' if infinite else args.games} "
f"start_id={args.start_id} max_moves={max_label} pass_recheck={args.pass_recheck} "
f"lessons h={lesson_book['to_heuristic']} r5={lesson_book['to_r5']}",
flush=True,
)
katago_counts: Counter = Counter()
judge: Optional[FinalJudge] = FinalJudge(args.katago_visits) if args.katago_judge else None
db: Optional[MatchDB] = None
tail: Optional[ServerLogTail] = None
if not args.no_db:
server_log = Path(args.server_log) if args.server_log else default_server_log(ROOT)
if server_log is not None and not server_log.is_absolute():
server_log = ROOT / server_log
db = MatchDB(out / "matches.sqlite")
tail = ServerLogTail(server_log)
run_id = db.start_run(_iso(), vars(args), HEURISTIC_VERSION, server_log)
print(f"[monitor] sqlite={out / 'matches.sqlite'} run={run_id} server_log={server_log}", flush=True)
gid = max(0, int(args.start_id))
while True:
gid += 1
if not infinite and gid > args.games:
break
print(f"[monitor] game {gid} …", flush=True)
rep = await play_one(
gid,
args.ws_url,
args.max_moves,
args.timeout_s,
args.seed,
pass_recheck=args.pass_recheck,
db=db,
tail=tail,
)
if judge is not None:
rep.katago_lead_b = judge.lead_black(rep.final_board)
if rep.score is not None and rep.katago_lead_b is not None:
rep.score["katago_lead_b"] = rep.katago_lead_b
rep.score["katago_winner"] = "black" if rep.katago_lead_b > 0 else "white"
katago_counts[rep.score["katago_winner"]] += 1
if db is not None:
db.finish_game(
gid,
ts=rep.finished_at,
end_reason=rep.end_reason,
plies=rep.plies,
score=rep.score,
issues=[asdict(i) for i in rep.issues],
katago_lead_b=rep.katago_lead_b,
)
end_counts[rep.end_reason] += 1
winner_counts[(rep.score or {}).get("winner") or "none"] += 1
for iss in rep.issues:
issue_counts[iss.kind] += 1
if rep.pass_checks:
with pass_cases_path.open("a", encoding="utf-8") as f:
for pc in rep.pass_checks:
recheck_counts[f"{pc['phase']}:{pc['result']}"] += 1
f.write(json.dumps({"game_id": gid, **pc}, ensure_ascii=False) + "\n")
_write_status(
out=out,
games_path=games_path,
issues_path=issues_path,
live_path=live_path,
gid=gid,
games_target=args.games,
rep=rep,
issue_counts=issue_counts,
end_counts=end_counts,
winner_counts=winner_counts,
recheck_counts=recheck_counts,
)
heart_path.write_text(
"\n".join(
[
f"# web monitor heartbeat — {_iso()}",
f"- games: `{gid}`",
f"- winners: `{dict(winner_counts)}`",
f"- katago_winners: `{dict(katago_counts)}`",
f"- pass_recheck: `{dict(recheck_counts)}`",
f"- early_vlm_pass: `{issue_counts.get('early_vlm_pass', 0)}`",
f"- early_double_pass: `{issue_counts.get('early_double_pass', 0)}`",
f"- vlm_illegal: `{issue_counts.get('vlm_illegal', 0)}`",
f"- vlm_comm: `{issue_counts.get('vlm_comm', 0)}`",
"",
]
),
encoding="utf-8",
)
winner = (rep.score or {}).get("winner")
if absorb_game(lesson_book, rep.moves_summary, winner, f"{out.name}:{rep.game_id}"):
save_book(lesson_book, LESSON_PATH)
print(
f"[monitor] lessons h={lesson_book['to_heuristic']} r5={lesson_book['to_r5']}",
flush=True,
)
if args.sleep_s > 0:
await asyncio.sleep(args.sleep_s)
if db is not None:
db.close()
print(f"[monitor] done → {issues_path}", flush=True)
return 0
def main() -> int:
p = argparse.ArgumentParser()
p.add_argument("--ws-url", default="ws://127.0.0.1:9320")
p.add_argument("--games", type=int, default=3, help="0 = infinite. 마지막 국 번호")
p.add_argument("--start-id", type=int, default=0, help="이 번호 다음 국부터 진행")
p.add_argument(
"--max-moves",
type=int,
default=MAX_GAME_PLIES,
help=f"0 = 수 제한 없음(쌍패스 종국, 안전 상한 {SAFETY_PLIES}수)",
)
p.add_argument(
"--pass-recheck",
action="store_true",
help="R5 PASS 시 같은 국면을 한 번 더 요청해 PASS 재현 여부를 기록",
)
p.add_argument("--timeout-s", type=float, default=240.0)
p.add_argument("--seed", type=int, default=42)
p.add_argument("--sleep-s", type=float, default=2.0, help="게임 사이 대기")
p.add_argument("--out-dir", default="logs/mariago_web_monitor")
p.add_argument(
"--server-log",
default="",
help="VLM 서버 --log-jsonl 경로 (기본 logs/vlm_eval_day/play_r5.jsonl). 백 수마다 같은 국면 레코드를 SQLite에 붙인다",
)
p.add_argument("--no-db", action="store_true", help="SQLite 기록 끔")
p.add_argument("--katago-judge", action="store_true", help="종국 판을 KataGo로 다시 계가(사석 반영)")
p.add_argument("--katago-visits", type=int, default=400)
args = p.parse_args()
return asyncio.run(amain(args))
if __name__ == "__main__":
raise SystemExit(main())