homeclaw/scripts/guard-firing-audit.py

#!/usr/bin/env python3
"""
가드 발동률 실측 — 2026-09-04 가드 패치(db6a863 / 10e54d3 / b2075c6) 이후 구간을 잰다.

왜 필요한가. 이번 패치들은 전부 실사용 로그를 표본으로 보정했지만, 보정이 맞았는지는
"패치 이후 로그"로만 확인할 수 있다([[feedback_calibrate_on_real_data]]). 특히 관련성
관문(sourceCoversQuery)은 양쪽으로 어긋날 수 있다:
  · 너무 헐거우면 → NUMERIC-GROUNDING이 계속 맞는 답을 지운다(원래 사고의 재발)
  · 너무 빡빡하면 → standing down이 과해져 진짜 날조가 그대로 통과한다

## 왜 줄 번호가 아니라 누적 카운트인가

로그는 크기 기준(1M)으로 회전하고 `copytruncate`를 쓴다 — 회전 시 내용이 .1로 복사되고
원본은 0바이트로 잘린다. 그래서 "몇 번째 줄 이후"라는 기준은 회전 한 번에 무효가 된다
(패치 당일 gateway.log가 이미 1.05M이라 그날 밤 바로 잘릴 참이었다).

대신 gateway.log* **전 파일(.gz 포함)** 을 합쳐 패턴별 누적 횟수를 세고, 기준선의 값을
뺀다. 회전해도 내용은 .1/.2.gz로 옮겨갈 뿐 사라지지 않으므로 이 방식은 회전에 견딘다.
`rotate 14`라 파일이 실제로 없어지려면 현재 증가 속도로 반년 이상 걸린다 — 그때는 차이가
음수로 나오고 스크립트가 경고한다.

사용:
    python3 scripts/guard-firing-audit.py            # 기준선 이후 증가분
    python3 scripts/guard-firing-audit.py --total    # 누적 전체(기준선 무시)
"""

import glob
import gzip
import json
import os
import re
import sys

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
LOG_GLOB = os.path.join(ROOT, '.smallclaw/logs/gateway.log*')
BASELINE = os.path.join(ROOT, '.smallclaw/logs/audit/guard-baseline.json')

# (키, 라벨, 정규식, 이번 패치와의 관계)
PATTERNS = [
    ('numeric_standdown',  'NUMERIC standing down', r'NUMERIC-GROUNDING: standing down', 'new'),
    ('numeric_reprompt',   'NUMERIC 재프롬프트',      r'NUMERIC-GROUNDING POST-CHECK', 'should-drop'),
    ('search_abandonment', 'SEARCH-ABANDONMENT',    r'SEARCH-ABANDONMENT POST-CHECK', 'should-rise'),
    ('dead_end_fallback',  'DEAD-END FALLBACK',     r'DEAD-END FALLBACK POST-CHECK', 'new'),
    ('sycophantic',        'SYCOPHANTIC',           r'SYCOPHANTIC-CONFIRMATION POST-CHECK', 'new'),
    ('auto_recover',       'AUTO-RECOVER',          r'\[v2\] AUTO-RECOVER \(', 'baseline'),
    ('empty_grounding',    'EMPTY-GROUNDING',       r'EMPTY-GROUNDING POST-CHECK', 'baseline'),
    ('intent_only',        'INTENT-ONLY',           r'INTENT-ONLY POST-CHECK', 'baseline'),
    ('messaging',          'MESSAGING',             r'MESSAGING POST-CHECK', 'baseline'),
    ('stale_version',      'STALE-VERSION QUERY',   r'STALE-VERSION QUERY', 'baseline'),
    ('dup_skip',           '중복 도구호출 SKIP',       r'SKIP: duplicate tool call', 'baseline'),
    ('synth_answer',       '합성 Answer: 줄',         r'TOOL OK: Answer: The current', 'should-be-0'),
    ('user_turns',         '사용자 턴',               r'^\[v2\] USER \[', 'baseline'),
]

NOTE = {
    'new': '이번에 추가',
    'should-drop': '줄어야 정상',
    'should-rise': '패치 전엔 이 어법을 못 잡았음',
    'should-be-0': '0이어야 정상',
    'baseline': '',
}

FIG_RE = re.compile(r'figure\(s\) not found in tool output \[([^\]]*)\]')
COV_RE = re.compile(r'cover only (\d+)% of the query')


def read_all_logs() -> str:
    parts = []
    for p in sorted(glob.glob(LOG_GLOB)):
        try:
            if p.endswith('.gz'):
                parts.append(gzip.open(p, 'rt', errors='replace').read())
            else:
                parts.append(open(p, errors='replace').read())
        except OSError as exc:
            print(f'  ⚠ {os.path.basename(p)} 읽기 실패: {exc}', file=sys.stderr)
    if not parts:
        sys.exit(f'로그를 찾지 못했습니다: {LOG_GLOB}')
    return '\n'.join(parts)


def main() -> None:
    text = read_all_logs()
    total = {key: len(re.findall(rx, text, re.M)) for key, _, rx, _ in PATTERNS}

    base = {}
    if '--total' not in sys.argv and os.path.exists(BASELINE):
        base = json.load(open(BASELINE)).get('counts', {})

    if base:
        delta = {k: total[k] - base.get(k, 0) for k in total}
        if any(v < 0 for v in delta.values()):
            print('⚠ 일부 카운트가 기준선보다 작습니다 — 오래된 로그 파일이 rotate 14를 넘겨')
            print('  삭제된 듯합니다. 아래 숫자는 신뢰하지 마시고 --total 로 보세요.\n')
        counts, label = delta, '기준선(2026-09-04) 이후 증가분'
    else:
        counts, label = total, '누적 전체'

    turns = counts.get('user_turns', 0)
    print(f'구간: {label} — 사용자 턴 {turns}건\n')
    if turns <= 0:
        print('아직 이 구간에 대화가 거의 없습니다. 며칠 더 쓰고 다시 도세요.')
        return

    print(f'{"가드":<24}{"발동":>6}{"턴당":>9}   비고')
    print('-' * 64)
    for key, lbl, _, kind in PATTERNS:
        if key == 'user_turns':
            continue
        c = counts[key]
        print(f'{lbl:<24}{c:>6}{c / turns:>9.3f}   {NOTE[kind]}')

    # 지워진 수치가 실제로 무엇이었는지 — 사용자가 원한 답이 지워졌는지는 눈으로만 판정된다.
    figs = FIG_RE.findall(text)
    if figs:
        print(f'\n지워진 수치 (누적 {len(figs)}건, 최근 12건):')
        for f in figs[-12:]:
            print(f'  · {f}')
        print('  ↑ 사용자가 원했던 답이 여기 섞여 있으면 관련성 관문이 아직 헐겁다.')

    covs = sorted(int(m) for m in COV_RE.findall(text))
    if covs:
        print(f'\nstanding down 커버리지: {covs}')
        print(f'  중앙값 {covs[len(covs) // 2]}%   (임계값 34%)')
        near = [c for c in covs if 25 <= c <= 33]
        if near:
            print(f'  ⚠ 임계값 바로 아래(25~33%) {len(near)}건 — 경계 사례가 쌓이면 값 재검토')

    print('\n판독 기준')
    print('  · 합성 Answer: 줄이 0이 아니면 → generic 경로가 되살아났는지 확인')
    print('  · SEARCH-ABANDONMENT이 여전히 0이면 → 포기 어법이 또 바뀐 것(NOTHING_FOUND 갱신)')
    print('  · DEAD-END FALLBACK이 턴당 0.05 넘으면 → PRIOR_DEAD_ENDS_REQUIRED=1이 낮은 것')
    print('  · standing down이 NUMERIC 재프롬프트보다 훨씬 많으면 → 관문이 너무 빡빡')
    print('  · 패치 전 기준(턴당): NUMERIC 재프롬프트 0.051 / AUTO-RECOVER 0.080 /')
    print('    SEARCH-ABANDONMENT 0.010 / 합성 Answer: 0.009')


if __name__ == '__main__':
    main()