# 작업지시서 — Phase B Verifier MVP (R1·R2·R4 tool-arg validators)

> 대상: 코딩 LLM. 이 문서만 보고 독립 수행 가능.
> 상위 플랜: `plans/OPUS-v2-실행계획.md` Phase B.
> 예상 분량: 새 모듈 1개(~150줄) + 기존 server.py 진입부에 데코레이터 적용. 로직 변경 없음.

## 1. 배경 (왜)

production opencode 채팅에서 Qwen3-8B(및 가끔 35B)가 **존재하지 않는 태그·잘못된 area 형식으로 MCP 도구를 호출**하고, 서버가 silent fallback해 모델이 그 결과를 사용자에게 잘못 포장하는 패턴 관찰:
- `trace_connections(start_tag="raw_material_input")` ← 합성 (영어로 만든 가짜 태그)
- `trace_connections(start_tag="RM-6101")` ← Raw Material + 실재 p-6101 차용 합성
- `generate_status_report(area="6-1")` ← `P6-1` 형식 무시 → 서버가 무필터 전체 결과 반환 → "6-1차 결과"로 둔갑

**Phase 0 eval은 통제 조건이라 이 패턴을 못 잡음.** Verifier = 코드 결정적 검증을 MCP 도구 입구에 박아, 잘못된 호출 즉시 reject + 자기교정 hint 반환 → 모델이 `find_tags` 먼저 호출하는 행동으로 전환.

부수효과: 거부 로그가 **Phase C(LoRA) 학습 데이터 자동 수집** 역할 (잘못된 호출 → hint → 올바른 호출 trip).

## 2. 수정 대상

| 파일 | 동작 |
|---|---|
| `mcp-server/verifier/__init__.py` | 신규 (빈 패키지 마커) |
| `mcp-server/verifier/validators.py` | 신규 — R1·R2 validator 함수 + VerifierError + 로그 적재 |
| `mcp-server/verifier/test_validators.py` | 신규 — 단위 테스트 (각 룰별 PASS/FAIL 2건씩 이상) |
| `mcp-server/verifier/logs/.gitkeep` | 신규 (빈 디렉토리 유지) |
| `mcp-server/verifier/README.md` | 신규 — 룰 카탈로그·로그 포맷 |
| `mcp-server/server.py` | **수정** — import + 적용 대상 tool 함수 입구에 검증 호출 (return 분기) |

⚠️ **수정 금지**: `mcp-server/worker/*`, `mcp-server/eval/*`, 채점기·골든셋·모델 서빙 스크립트.

## 3. 룰 명세 (정확히 이대로)

### R1 — `tag-existence` (모든 tool 인자의 base_tag)
- 입력 패턴: 정규식 `^[a-z][a-z0-9]*-\d+[a-z]?(\.[a-z0-9]+)?$` (예: `ficq-6113`, `ficq-6113.pv`, `p-6102`, `vp-6117a`)
- 검증: `.` 앞부분(base_tag)이 `tag_metadata.base_tag` ∪ `pid_equipment.tag_no` (소문자 비교)에 존재해야 PASS
- 캐시: 모듈 전역 `set`, TTL **300초**(5분). 캐시 미스 시 두 테이블 한번에 로드
- 실패 시 hint에 **부분 매칭 top 3 suggested** 첨부 (split('-') 토큰 길이 2 초과만)

### R2 — `area-format` (`area`, `sub_area` 인자)
- 정규식: `^P\d+(-\d+)?$` (예: `P6`, `P6-1`). 빈 문자열/`None`은 PASS (선택 필터)
- 추가: `-` 앞 부분이 유효 area 코드여야 함:
  `{"P1","P2","P3","P4","P5","P6","P8","P9","P10","UTIL","PACKING"}` (P7 없음)
- 실패 hint: 형식 또는 미존재 모두 valid 코드 목록 포함

### R4 — `trace_connections` 보강
- `start_tag`: R1 적용 (위와 동일)
- `direction`: `{"upstream", "downstream"}` 외 reject
- `max_depth`: int, 1~50 외 reject

> ⚠️ **R3·R5는 이번 MVP 범위 밖** (응답 텍스트 후처리/LLM-judge 필요). `validators.py`에 stub 함수만 두고 `NotImplementedError` 또는 NO-OP로 표시. Phase B.2에서 구현.

## 4. 구현 골자 (정확히 따를 것)

### 4.1 `mcp-server/verifier/validators.py`

```python
"""Phase B Verifier MVP — tool 인자 결정적 검증.

룰: R1(tag-existence), R2(area-format), R4(trace_connections 보강).
응답 텍스트 검증(R3, R5)은 Phase B.2 — stub만.
"""
from __future__ import annotations
import json, re, time, pathlib
from typing import Optional, Any

TAG_RE  = re.compile(r'^[a-z][a-z0-9]*-\d+[a-z]?(\.[a-z0-9]+)?$')
AREA_RE = re.compile(r'^P\d+(-\d+)?$')
VALID_AREAS = {"P1","P2","P3","P4","P5","P6","P8","P9","P10","UTIL","PACKING"}
VALID_DIRECTIONS = {"upstream","downstream"}

_LOG_DIR = pathlib.Path(__file__).parent / "logs"

class VerifierError(Exception):
    def __init__(self, rule: str, code: str, hint: str, **extra):
        self.rule, self.code, self.hint, self.extra = rule, code, hint, extra
    def to_dict(self) -> dict:
        return {"verifier_error": f"{self.rule}.{self.code}", "hint": self.hint, **self.extra}

# ── 태그 캐시 ──
_tag_cache: set[str] | None = None
_tag_cache_at: float = 0.0

def _load_tag_set(get_conn) -> set[str]:
    global _tag_cache, _tag_cache_at
    if _tag_cache is not None and (time.time() - _tag_cache_at) < 300:
        return _tag_cache
    conn = get_conn()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT DISTINCT base_tag FROM tag_metadata WHERE base_tag IS NOT NULL")
            s = {r[0].lower() for r in cur.fetchall() if r[0]}
            cur.execute("SELECT DISTINCT tag_no FROM pid_equipment WHERE tag_no IS NOT NULL")
            s |= {r[0].lower() for r in cur.fetchall() if r[0]}
    finally:
        conn.close()
    _tag_cache, _tag_cache_at = s, time.time()
    return s

# ── R1 ──
def validate_tag(tag: str | None, get_conn) -> Optional[VerifierError]:
    if not tag:
        return None
    t = tag.lower()
    if not TAG_RE.match(t):
        return VerifierError("R1","invalid_tag_format",
            hint=f"태그 형식 비정상: '{tag}'. 예시: ficq-6113.pv, p-6102")
    base = t.split('.')[0]
    tags = _load_tag_set(get_conn)
    if base in tags:
        return None
    toks = [p for p in base.split('-') if len(p) > 2]
    suggested = sorted({x for x in tags if any(p in x for p in toks)})[:3]
    return VerifierError("R1","tag_not_found",
        hint=f"태그 '{tag}' 는 DB에 존재하지 않습니다. find_tags(query=..., sub_area=...) 로 먼저 검색하세요.",
        suggested=suggested)

# ── R2 ──
def validate_area(area: str | None, field: str = "area") -> Optional[VerifierError]:
    if not area:
        return None
    if not AREA_RE.match(area):
        return VerifierError("R2","invalid_area_format",
            hint=f"{field}='{area}' 형식 오류. 'P6' 또는 'P6-1' 형식 사용.",
            valid_areas=sorted(VALID_AREAS))
    base = area.split('-')[0]
    if base not in VALID_AREAS:
        return VerifierError("R2","unknown_area",
            hint=f"{field}='{area}' 미존재. valid: {sorted(VALID_AREAS)} (P7 없음)")
    return None

# ── R4 ──
def validate_direction(d: str | None) -> Optional[VerifierError]:
    if d and d not in VALID_DIRECTIONS:
        return VerifierError("R4","invalid_direction",
            hint=f"direction='{d}' 잘못. 'upstream' 또는 'downstream' 만 허용")
    return None

def validate_max_depth(n: Any) -> Optional[VerifierError]:
    if n is None: return None
    try:
        v = int(n)
    except Exception:
        return VerifierError("R4","invalid_max_depth", hint=f"max_depth='{n}' 은 정수여야 함")
    if not (1 <= v <= 50):
        return VerifierError("R4","max_depth_out_of_range", hint=f"max_depth={v} 범위 외 (1~50)")
    return None

# ── R3, R5 stub (Phase B.2) ──
def validate_response_text(text: str) -> Optional[VerifierError]:
    return None  # Phase B.2 구현 예정

# ── 로그 적재 (Phase C LoRA 입력) ──
def log_rejection(tool: str, params: dict, err: VerifierError) -> None:
    _LOG_DIR.mkdir(parents=True, exist_ok=True)
    today = time.strftime("%Y-%m-%d")
    rec = {"ts": time.time(), "tool": tool, "params": params,
           "verifier_error": err.to_dict()}
    with (_LOG_DIR / f"{today}.jsonl").open("a", encoding="utf-8") as f:
        f.write(json.dumps(rec, ensure_ascii=False) + "\n")
```

### 4.2 server.py 적용 — 각 tool 입구에 한 줄

DB 커넥션은 server.py에 이미 있는 헬퍼(예: `_get_db_connection`)를 `get_conn` 자리에 전달.

```python
from verifier.validators import (
    validate_tag, validate_area, validate_direction, validate_max_depth,
    log_rejection, VerifierError,
)

def _check(tool: str, params: dict, *errs) -> dict | None:
    """첫 번째 비-None 에러를 로그 + 반환(dict). 없으면 None."""
    for e in errs:
        if e:
            log_rejection(tool, params, e)
            return e.to_dict()
    return None
```

**적용 대상 (server.py의 기존 함수 시그너처는 그대로 두고, 본문 첫 줄에 `_check` 호출)**:

| Tool | 검증 |
|---|---|
| `find_tags` | `validate_area(area)`, `validate_area(sub_area, "sub_area")` |
| `active_alarms` | `validate_area(area)` |
| `query_events` | `validate_area(area)` + `tag_name`이 인자면 `validate_tag` |
| `summarize_events` | `validate_area(area)` |
| `generate_status_report` | `validate_area(area)` |
| `query_pv_history` | `tag_names` list 각각 `validate_tag` (첫 실패 반환) |
| `trace_connections` | `validate_tag(start_tag, conn)`, `validate_direction(direction)`, `validate_max_depth(max_depth)` |
| `upsert_pid_connection` | `validate_tag(tag_no)` |

> tool이 dict를 반환하는 경우(이미 그렇게 되어있음) 그대로 `return _check(...) or 기존_본문()` 패턴. tool이 str을 반환하면 dict를 json.dumps로 감싸서 일관 유지.

### 4.3 `test_validators.py` (필수, 최소 10케이스)

각 룰별 PASS 2 + FAIL 2 이상. 예:
```python
def test_R1_valid_tag(monkeypatch):
    monkeypatch.setattr(validators, "_load_tag_set", lambda gc: {"ficq-6113","p-6102"})
    assert validators.validate_tag("ficq-6113.pv", lambda: None) is None
def test_R1_unknown_tag(monkeypatch):
    monkeypatch.setattr(validators, "_load_tag_set", lambda gc: {"ficq-6113"})
    err = validators.validate_tag("rm-6101", lambda: None)
    assert err and err.code == "tag_not_found"
def test_R2_invalid_area():
    err = validators.validate_area("6-1")
    assert err and err.code == "invalid_area_format"
# ... R2 PASS, R4 direction/max_depth, etc.
```

### 4.4 README.md (간단)

룰 카탈로그 표 + 로그 jsonl 포맷 예시 1줄 + Phase B.2(R3/R5) 미구현 명시.

## 5. 검증 (순서대로)

```bash
cd mcp-server

# 1) 컴파일
python3 -m py_compile verifier/validators.py server.py

# 2) 단위 테스트
.venv/bin/python -m pytest verifier/test_validators.py -v

# 3) MCP 서버 재시작
# (사용자 환경: experioncrawler systemd 또는 uv run server.py)

# 4) opencode 재현 시도 — 다음 3개 호출 시 verifier_error 반환되는지
#    a) trace_connections(start_tag="raw_material_input", direction="downstream") → R1 tag_not_found
#    b) trace_connections(start_tag="RM-6101", ...)                                → R1 tag_not_found + suggested 포함
#    c) generate_status_report(area="6-1", hours=24)                                → R2 invalid_area_format
#    각 응답에 verifier_error.{rule, code, hint} 키 확인

# 5) 로그 적재
ls -la verifier/logs/
cat verifier/logs/$(date +%Y-%m-%d).jsonl | head -3  # 거부 3건 보여야 함

# 6) 정상 호출 회귀 0건 확인
#    a) trace_connections(start_tag="p-6102", direction="upstream") → 정상 결과
#    b) generate_status_report(area="P6-1", hours=24)                → 정상 결과
#    c) Phase 0 eval 재실행 (옵션) — 회귀 0
cd eval && ../.venv/bin/python run_eval.py --model Qwen3-8B --no-think \
  --baseline results/Qwen3-8B_20260526_103459.json
```

## 6. 수용 기준

- [ ] `py_compile` 통과 (validators.py, server.py)
- [ ] 단위 테스트 ≥ 10케이스 전부 PASS
- [ ] opencode 재현 §5.4 a/b/c **3건 모두 verifier_error 반환** (hint·suggested 포함)
- [ ] 정상 호출 §5.6 a/b 정상 결과 (회귀 없음)
- [ ] Phase 0 eval `--baseline` 비교에서 PASS→FAIL 회귀 0건
- [ ] `verifier/logs/YYYY-MM-DD.jsonl` 에 거부 3건 자동 적재 확인

## 7. 하지 말 것 (금지)

- ❌ 응답 텍스트(LLM 출력) 후처리 — Phase B.2 범위 (R3/R5는 stub만)
- ❌ LLM-judge 사용 (MVP는 코드 결정적만)
- ❌ tool 함수 시그너처·반환 타입 변경 (입구에 한 줄 추가만)
- ❌ 골든셋·평가 러너·채점기 변경
- ❌ 모델 서빙 설정·chat 템플릿 변경
- ❌ 캐시 TTL·매직 넘버 튜닝 (300초·top3·1~50 고정)
- ❌ 새 DB 객체(테이블/뷰) 생성 (read-only 검증)
- ❌ `worker/` 디렉토리 변경

## 8. 산출물

- 신규 6개: `verifier/{__init__.py, validators.py, test_validators.py, README.md, logs/.gitkeep}`
- 수정 1개: `mcp-server/server.py` (import + tool 8개 입구 `_check` 호출)
- 한 줄 보고: 적용된 tool 개수 / 재현 §5.4 결과 (3/3) / 회귀 (0) / 로그 샘플 1행