Yun Chan 56a6e2da93 chore: 저장소 구조 정리 및 문서화, 첫 커밋

- src/dist 산출물 분리 원칙 정리(.gitignore, .gitattributes)
- 루트 및 주요 폴더(config/scripts/prompts/tests/src, 런타임 폴더 5종)에
  안내용 README.md 추가
- CHANGELOG.md, LICENSE, docs/ops/05-release-and-versioning.md 추가
- docs/README.md 문서 지도 갱신

2026-09-04 09:25:44 +09:00

230 KiB

Raw Permalink Blame History

장애 알림과 복구 안내 설계

이 문서의 역할: DMF Crawler 가 실패했을 때 사람이 반드시 알아채고, 그 자리에서 고칠 수 있게 만드는 알림·복구 계층의 정본(SSOT)이다. 알림 채널 선택과 그 실패 조건, 4등급 체계와 폭주 억제, 상황별 실제 문구 전집, 버튼 액션 구현 코드, agy 재로그인 유도 경로, 알림 스크립트 전문, 테스트 절차, 에스컬레이션 규칙까지 이 문서 하나로 구현·검증이 끝나야 한다. 상위 정본은 docs/design/01-architecture.md(ADR-10/11/12, §3.14 alerts.py, §3.16 notify/pump.py, §7 실패 시나리오 대응표)이고, 운영 환경 근거는 docs/research/08-windows-scheduling-and-resilience.md, agy 사실관계는 docs/research/05a-agy-cli-ssot.md 다.

0. 한눈에 보기

이 문서가 확정하는 것:

알림 채널은 4단 사다리다. ① tkinter 토스트(우하단 자동소멸) → ② tkinter 강제 모달(복구 GUI) → ③ Windows 이벤트 로그 + state/alerts.json(항상, 무조건) → ④ 웹훅(선택, CRITICAL 전용). BurntToast·win11toast·pywin32는 쓰지 않는다(ADR-12 재확인). 외부 모듈 설치가 전제인 알림기는 "설치가 깨지면 알림도 안 뜨는" 순환 실패를 만든다.
토스트가 안 뜨는 조건은 5가지이고, 전부 대응이 있다: 로그온 전 / Session 0(S4U 배치) / 집중 지원(방해 금지) / 전체 화면 앱 / GUI 스택 자체 손상. 대응의 뼈대는 하나 — 알림을 발생시키는 프로세스와 표시하는 프로세스를 분리하고, 발생 기록은 절대 유실되지 않게 3중으로 남긴다(ADR-11).
등급은 4개다: INFO / WARN / ERROR / CRITICAL. 아키텍처 §3.14의 Severity 3값(INFO/WARN/CRITICAL)에 ERROR 를 추가한다(AMD-01). 판정 기준은 단 하나 — 오늘 xlsx 리포트가 나왔는가. 나왔으면 최대 WARN, 안 나왔으면 ERROR, 여기에 "반복" 또는 "사람이 손대야만 풀린다"가 붙으면 CRITICAL.
폭주 억제는 3중이다: ① dedup_key(코드+실행일자) + 코드별 쿨다운, ② 같은 pump 주기 내 다건 병합(N건을 토스트 1장으로), ③ 시간당 토스트 상한·일일 알림 상한. CRITICAL 모달만 상한을 면제받되 스누즈(기본 60분)를 갖는다.
문구는 문자열이 아니라 계약이다. 모든 알림은 [무엇][왜][어떻게][다음 행동] 4요소를 반드시 갖고, 하나라도 비면 raise_alert()가 ValueError를 던진다. 본 문서 §3에 22개 시나리오의 완성 문구 전집을 싣는다.
버튼 액션은 액션 키 레지스트리로 관리한다. open_log_dir / run_now / agy_relogin / install_agy / enter_api_key / reregister_tasks / restore_backup / open_report_dir / cleanup_disk / open_doctor / snooze / dismiss 12종. 액션 없는 실패 알림은 등록 자체가 거부된다(막다른 골목 금지, R7.7).
dmf:// 프로토콜 핸들러는 기본 경로에서 불필요하다. tkinter 버튼이 같은 프로세스 안에서 함수를 직접 호출하기 때문이다. 다만 진짜 Windows 토스트(액션 센터 잔류)를 선택 의존성으로 켜는 날을 위해 등록 절차를 §4.4에 완비해 둔다.
agy 재로그인은 "새 콘솔 창"으로만 가능하다. S4U 배치 세션에는 데스크톱이 없어 OAuth 브라우저가 뜨지 않는다. 로그온 세션의 pump 가 cmd.exe /c start 로 보이는 콘솔을 띄우고, 사용자가 로그인을 마치면 pump 가 헬스 프롬프트 1회로 성공을 검증한다.
scripts/notify.ps1 을 신설한다(아키텍처 트리 증분, AMD-02). Python/venv 가 통째로 깨졌을 때(실패 시나리오 #32) tkinter 토스트는 원리적으로 뜰 수 없다. PowerShell 단독으로 도는 최후 알림기가 Agent 작업의 두 번째 액션으로 등록되어, pump 가 스탬프를 남기지 못했을 때만 MessageBox 를 띄운다.
에스컬레이션은 3일/5일/7일 3단이다. 연속 실패 3일 → CRITICAL 모달 + 진단 자동 표시, 5일 → 웹훅 강제 발사(설정돼 있으면), 7일 → 배치 자동 실행 중단(state/paused.flag) 후 사람의 명시적 재개를 요구한다. 고장난 배치가 매일 조용히 실패하며 API 를 두드리는 상태를 방치하지 않는다.

1. 알림 채널 확정

1.1 채널 목록과 채택 근거

#	채널	구현	프로세스	언제 쓰는가	채택
C1	자동소멸 토스트	`notify/toast.py` (tkinter `overrideredirect` 창)	Agent(Interactive)	INFO·WARN. 12초 후 사라짐. 무시해도 되는 알림	✅ 기본
C2	강제 모달 복구 창	`gui/app.py` (tkinter `Toplevel` + `grab_set`)	Agent(Interactive)	ERROR·CRITICAL. 사용자가 액션을 고를 때까지 유지	✅ 기본
C3	Windows 이벤트 로그	`notify/eventlog.py` (`eventcreate.exe`)	배치·Agent 양쪽	전 등급 무조건. 화면이 없어도 남는 유일한 OS 표준 흔적	✅ 기본
C4	상태 파일 미러	`alerts.mirror_to_file()` → `state/alerts.json`	배치	전 등급 무조건. Agent 가 SQLite 잠금 없이 읽는 경로	✅ 기본
C5	MessageBox 폴백	`scripts/notify.ps1` (`System.Windows.Forms.MessageBox`)	별도 PowerShell	Python 이 깨져 C1·C2 가 불가능할 때	✅ 폴백
C6	`msg.exe` 세션 메시지	`scripts/notify.ps1` 내부	별도 PowerShell	.NET 로드조차 실패할 때의 최후 수단	✅ 최후
C7	웹훅(Discord/Slack/Telegram/generic)	`notify/webhook.py`	배치 또는 Agent	CRITICAL·에스컬레이션. PC 앞에 사람이 없을 때 닿는 유일한 채널	⚪ 선택(기본 off)
C8	이메일(SMTP)	—	—	웹훅으로 대체	❌ 기각
C9	BurntToast / win11toast	PowerShell 모듈 / pip 패키지	—	—	❌ 기각(ADR-12)
C10	dead-man switch(healthchecks.io 등)	`notify/webhook.py` 의 ping 모드	배치	PC 가 통째로 꺼져 있는 경우 감지	⚪ 선택(기본 off)

C8(이메일) 기각 근거: SMTP 는 앱 비밀번호·포트 차단·2FA 정책이라는 실패 표면 3개를 추가하는데, 웹훅은 URL 하나면 되고 실패해도 HTTP 상태 코드로 즉시 진단된다. 요구 N5(의존성 최소)와 R6.3(비개발자가 설정)을 동시에 만족하는 쪽은 웹훅이다.

C9(BurntToast/win11toast) 기각 근거 재확인: ADR-12 가 이미 기각했다. 추가로, 알림 채널이 외부 모듈에 의존하면 실패 시나리오 #32(Python/venv 손상)에서 알림 자체가 침묵한다. 알림기는 시스템에서 가장 의존성이 적어야 하는 컴포넌트다. tkinter 는 CPython 표준 배포에 포함되고, eventcreate.exe·msg.exe·powershell.exe 는 Windows 에 내장된다.

1.2 토스트가 안 뜨는 조건과 대응

#	조건	왜 안 뜨는가	감지 방법	대응
B1	로그온 전 / 로그오프 상태	표시할 대화형 세션이 존재하지 않는다	Agent 작업이 아예 실행되지 않음	배치는 `alerts` + `state/alerts.json` + 이벤트 로그에 축적. Agent 작업에 `AtLogOn` 트리거를 걸어 다음 로그온 즉시 밀린 알림을 병합 표시(시나리오 #31)
B2	Session 0 격리(S4U 배치 세션)	Windows Vista 이후 서비스·비대화형 세션은 사용자 데스크톱과 분리된다. tkinter 창을 만들어도 아무도 못 본다	`pipeline` 은 애초에 UI 를 호출하지 않는다(ADR-11)	구조로 해결. 배치는 알림 의도만 기록하고, Interactive 로 도는 Agent 가 표시한다. 최대 지연 15분(`schedule.agent_repeat_minutes`)
B3	집중 지원 / 방해 금지 모드	OS 알림 센터가 억제한다	우리 토스트는 OS 알림 API 를 쓰지 않는 자체 tkinter 창이므로 영향받지 않는다	해당 없음. 이것이 tkinter 를 택한 부수 이득이다. 단, 전체 화면 앱 위에서는 B4 로 넘어간다
B4	전체 화면 앱(게임·프레젠테이션)	우리 창이 `-topmost` 여도 배타적 전체 화면 앞에는 못 온다	`ctypes` 로 `SHQueryUserNotificationState` 조회	`QUNS_BUSY`/`QUNS_RUNNING_D3D_FULL_SCREEN`/`QUNS_PRESENTATION_MODE` 이면 표시를 미루고 다음 주기에 재시도. CRITICAL 은 미루지 않고 표시하되 이벤트 로그·웹훅을 함께 발사
B5	AppId(AUMID) 미등록	진짜 Windows 토스트는 등록된 AUMID 없이는 XML 토스트를 띄울 수 없다	해당 없음	우리 경로에는 해당하지 않는다. tkinter 창은 AUMID 를 요구하지 않는다. 선택 의존성으로 win11toast 를 켜는 날에만 §4.4 의 프로토콜/AUMID 등록이 필요해진다
B6	tkinter 자체 손상 / Python 실행 불가	`import tkinter` 실패, venv 파괴	Agent 액션이 비0으로 죽고 `state/pump.stamp` 가 갱신되지 않음	C5 폴백. Agent 작업의 2번째 액션 `scripts/notify.ps1 -Mode Guard` 가 스탬프 나이를 보고 MessageBox 를 띄운다(§6.1)
B7	Agent 작업 자체가 미등록/비활성	사용자가 지웠거나 정책이 껐다	`checks` ⑨ 가 `Get-ScheduledTask` 로 확인	배치가 `TASK_MISSING` CRITICAL 을 기록. 다만 표시할 주체가 없으므로 웹훅과 이벤트 로그가 유일한 통로 → 웹훅을 켜라고 온보딩에서 유도

핵심: B1~B7 중 어느 하나가 걸려도 알림이 사라지지는 않는다. alerts 테이블은 append-only 이고 shown_at IS NULL 인 행은 표시될 때까지 계속 대기한다. "표시 실패"는 지연일 뿐 유실이 아니다.

1.3 폴백 사다리 (표시 경로 결정)

알림 발생 (배치 또는 Agent 내부 워치독)
  │
  ├─[항상] alerts 테이블 INSERT ────────────────┐
  ├─[항상] state/alerts.json 미러 갱신 ─────────┤ 유실 방지 3중 기록
  └─[항상] Windows 이벤트 로그 기록 ────────────┘
              │
              ▼
   Agent(Interactive, 15분 주기 + AtLogOn) 가 pending() 조회
              │
   ┌──────────┴───────────────────────────────┐
   │ 등급 판정                                  │
   ├─ INFO / WARN  → C1 토스트(12초 자동소멸)   │
   │     └ 실패 시 → 다음 주기 재시도(최대 3회) │
   │        └ 3회 실패 → ERROR 로 승격 → C2     │
   ├─ ERROR / CRITICAL → C2 강제 모달(복구 GUI) │
   │     └ GUI 기동 실패 → C5 MessageBox        │
   │          └ .NET 실패 → C6 msg.exe          │
   └────────────────────────────────────────────┘
              │
   [CRITICAL 이고 webhook_enabled] → C7 웹훅 발사(표시 성공 여부와 무관하게)
              │
   표시 성공 → alerts.mark_shown(alert_id) → state/alerts.json 재미러

규칙 3개

기록과 표시는 분리된다. 표시 실패는 기록을 되돌리지 않는다.
한 단계 아래로만 떨어진다. C1 실패가 곧바로 C6 로 가지 않는다. 단계마다 이벤트 로그에 NOTIFY_DEGRADED 를 남겨 어느 단에서 떨어졌는지 사후 추적이 된다.
웹훅은 폴백이 아니라 병렬 채널이다. CRITICAL 은 화면 표시 성공 여부와 무관하게 웹훅을 쏜다. 화면을 본 사람과 웹훅을 받는 사람이 같은 사람이라도, "PC 앞에 있었는가"는 알 수 없기 때문이다.

1.4 웹훅 채널 정책

항목	확정
기본 상태	off (`notify.webhook_enabled = false`). 온보딩 GUI 의 선택 단계에서 켠다
발사 조건	`severity >= notify.webhook_min_severity`(기본 `"CRITICAL"`) 또는 에스컬레이션 5일차 이상
지원 형식	`discord` / `slack` / `telegram` / `generic`(JSON POST) — `notify.webhook_kind` 로 선택
URL 보관	평문 금지. `secrets_dpapi.save("webhook_url", url)` 로 DPAPI 암호화. `config.toml` 에는 `webhook_kind` 와 on/off 만 들어간다
타임아웃	`notify.webhook_timeout_seconds` 기본 10초. 실패해도 절대 예외를 전파하지 않는다(알림기가 파이프라인을 죽이면 안 된다)
재시도	1회만. 웹훅 실패는 이벤트 로그에 `NOTIFY_DEGRADED` 로만 남긴다
내용	4요소 전문 + `run_id` + 로그 디렉터리 절대경로. API 키·토큰·URL 파라미터는 절대 포함하지 않는다(`logging.mask_patterns` 를 웹훅 본문에도 적용)
dead-man switch	`notify.deadman_url` 이 설정되면 배치 성공 시 루트 URL, 실패 시 `/fail` 을 GET. PC 가 꺼져 있으면 상대편이 알아챈다

1.5 채널 라우팅 매트릭스

등급	C1 토스트	C2 모달	C3 이벤트 로그	C4 alerts.json	C7 웹훅	재표시
INFO	선택(`notify.show_info_toast`, 기본 false)	✕	✅	✅	✕	안 함
WARN	✅ 12초	✕	✅	✅	✕	쿨다운 후 1회
ERROR	✕	✅ 모달	✅	✅	설정 시	60분마다
CRITICAL	✕	✅ 모달 + 포커스 강제	✅	✅	✅(켜져 있으면)	60분마다, 해소까지

2. 알림 등급 체계

2.1 4등급 정의 — AMD-01 (아키텍처 §3.14 개정)

아키텍처 §3.14 의 Severity 는 INFO / WARN / CRITICAL 3값이다. 이 문서는 ERROR 를 추가해 4값으로 개정한다.

개정 사유: 3값 체계에서는 "리포트가 안 나왔다"(오늘 산출물 부재)와 "인증이 만료됐다"(사람이 손대야만 풀림)가 둘 다 CRITICAL 로 뭉개진다. 그런데 두 상황의 필요한 사용자 행동이 다르다 — 전자는 "지금 다시 실행" 한 번이면 풀릴 수 있고, 후자는 반드시 사람이 계정 작업을 해야 한다. 채널·재표시 주기·에스컬레이션 카운터가 전부 갈리므로 등급을 분리하는 편이 싸다.

# src/dmf_crawler/alerts.py (개정)
class Severity(StrEnum):
    INFO     = "INFO"
    WARN     = "WARN"
    ERROR    = "ERROR"
    CRITICAL = "CRITICAL"

    @property
    def rank(self) -> int:
        return {"INFO": 0, "WARN": 1, "ERROR": 2, "CRITICAL": 3}[self.value]

    def at_least(self, other: "Severity") -> bool:
        return self.rank >= other.rank

판정 기준 — 질문 두 개로 끝난다.

Q1. 오늘 xlsx 리포트 파일이 존재하는가?
     +- 예   -> Q2 로
     +- 아니오 -> ERROR (최소)
Q2. 사람이 개입해야만 풀리는가? / 이미 반복되고 있는가?
     +- 예   -> CRITICAL
     +- 아니오 -> 데이터 품질에 영향이 있으면 WARN, 없으면 INFO

등급	정의	실행 상태	종료 코드	사용자에게 요구하는 것
`INFO`	정상 완료, 또는 자동 회복된 일시 장애	`SUCCESS`	0	아무것도. 기록만 남는다
`WARN`	부분 실패. 리포트는 생성됐다. 데이터가 스테일하거나 일부 계층이 빠졌다	`SUCCESS` 또는 `PARTIAL`	0	인지. 오늘 리포트를 볼 때 배너를 확인
`ERROR`	리포트 생성 실패. 오늘 산출물이 없다. 일시적 원인일 수 있어 재시도 가치가 있다	`FAILED`	1	지금 조치. "다시 실행" 한 번으로 풀릴 수 있다
`CRITICAL`	전제조건 붕괴(인증·키·DB·스케줄) 또는 연속 실패. 재시도해도 같은 결과	`FAILED` / `BLOCKED`	1 또는 2	반드시 사람이 손을 대야 한다. 모달로 막는다

2.2 등급별 채널·표시 방식·빈도

등급	표시	지속	최초 표시 지연	재표시 주기	쿨다운(동일 코드)	웹훅
INFO	(기본 미표시)	—	—	없음	1440분	✕
WARN	우하단 토스트	`notify.toast_seconds` 12초 자동소멸	최대 15분	없음(쿨다운 만료 시 1회)	`notify.cooldown_minutes` 240분	✕
ERROR	강제 모달(복구 GUI, `recover` 모드)	사용자가 닫을 때까지	최대 15분	`notify.modal_repeat_minutes` 60분	60분	설정 시
CRITICAL	강제 모달 + `focus_key` 로 해당 체크 항목 강조 + 창 포커스 강제	사용자가 닫을 때까지	최대 15분	60분	60분	켜져 있으면 ✅

표시 지연이 최대 15분인 이유: Agent 작업의 반복 주기(schedule.agent_repeat_minutes 기본 15)다. 요구 N3("24시간 내 인지")에 대해 96배의 여유가 있으므로 더 짧게 만들 이유가 없다. 주기를 줄이면 로그온 세션에서 도는 프로세스 기동 횟수만 늘어난다.

2.3 `dedup_key` 와 쿨다운

dedup_key 산출 규칙

def make_dedup_key(code: str, run_date: str | None, scope: str | None = None) -> str:
    """
    code      : 알림 코드 (AGY_AUTH, REPORT_LOCKED 등). 대문자 스네이크.
    run_date  : YYYY-MM-DD. 하루 단위로 재발생을 허용할 알림만 넣는다.
    scope     : 같은 코드라도 대상이 다르면 별개 알림인 경우의 구분자
                (예: 디스크 부족의 드라이브 문자 'D:', 잠긴 파일명).
    """
    parts = [code]
    if run_date:
        parts.append(run_date)
    if scope:
        parts.append(scope)
    return "|".join(parts)

코드 유형	`run_date` 포함?	이유
일자성 실패(`FETCH_FAILED`, `INTEGRITY_BLOCKED`, `REPORT_LOCKED`, `API_QUOTA_EXCEEDED`)	포함	어제 발생했던 것이 오늘 또 발생하면 새 사건이다. 알려야 한다
상태성 실패(`AGY_AUTH`, `API_KEY_MISSING`, `DB_CORRUPT`, `TASK_MISSING`, `AGY_MISSING`)	미포함	해소될 때까지 하나의 사건이다. 매일 새 알림을 만들면 목록이 오염된다. 재촉은 `modal_repeat_minutes` 가 담당
누적성(`CONSECUTIVE_FAILURES`, `WATCHDOG_STALE`)	미포함	위와 동일. 다만 본문의 "N일째"가 갱신된다

쿨다운 동작

raise_alert(code=X) 호출
  |
  +- 같은 dedup_key 의 미해소 행이 있는가?
       +- 없다 -> INSERT, occurrences=1, 반환 True (표시 대상)
       +- 있다 -> 마지막 발생 시각으로부터 cooldown 이 지났는가?
                   +- 안 지남 -> occurrences += 1, last_seen_at 갱신,
                   |             페이로드만 최신으로 UPDATE, 반환 False (표시 안 함)
                   +- 지남   -> occurrences += 1, last_seen_at 갱신,
                                 shown_at = NULL 로 되돌림, 반환 True (재표시)

alerts 는 append-only 인데 UPDATE 를 하는가? — AMD-03 아키텍처 §3.9 불변식은 "alerts 를 UPDATE/DELETE 하지 않는다"고 못박았다. 이 문서는 그 불변식을 깨지 않기 위해 테이블을 둘로 나눈다. 발생 사실은 alert_events(순수 append-only, 매 발생마다 1행), 표시·해소 상태는 alerts(dedup_key 유니크, 상태 머신)다. occurrences·last_seen_at·shown_at·resolved_at 은 상태이지 사실이 아니다. 감사 추적은 alert_events 가 온전히 보존한다.

2.4 알림 폭주 억제 3중 방어

층	이름	대상	동작	설정 키
L1	코드별 쿨다운	같은 코드의 반복	2.3 참조. 쿨다운 안이면 카운터만 증가	`notify.cooldown_minutes`(WARN 240분), `notify.modal_repeat_minutes`(ERROR/CRITICAL 60분)
L2	주기 내 병합	서로 다른 코드가 동시에 여러 개	한 pump 주기에서 표시 대상이 2건 이상이면 토스트 1장에 요약하고 "자세히 보기"로 GUI 를 연다	`notify.merge_threshold`(기본 2)
L3	총량 상한	하루 전체	시간당 토스트 `notify.max_toasts_per_hour`(6), 일일 알림 표시 `notify.daily_alert_cap`(20) 초과 시 그날은 표시를 멈추고 `NOTIFY_SUPPRESSED` INFO 만 기록	위 두 키

L3 의 CRITICAL 면제: CRITICAL 은 상한을 적용받지 않는다. 대신 스누즈(notify.snooze_minutes 기본 60)가 있어 사용자가 "나중에"를 누르면 그 시간만큼 조용해진다. 상한으로 CRITICAL 을 막으면 가장 중요한 알림이 가장 먼저 침묵하는 역설이 생긴다.

L2 병합 문구 실제 예시

제목: DMF 크롤러 — 확인이 필요한 항목 3건
본문: 2026-09-02 06:04 실행에서 확인할 항목이 3건 있습니다.
      · [주의] 수집 건수가 어제보다 6.2% 줄어 비교를 건너뛰었습니다
      · [주의] 리포트를 다른 이름으로 저장했습니다 (원본이 열려 있음)
      · [주의] AI 요약을 건너뛰었습니다 (일일 쿼터 소진)
      오늘 리포트는 정상 생성됐습니다.
버튼: [자세히 보기]  [리포트 열기]  [닫기]

2.5 승격 규칙과 해소 규칙

규칙	조건	결과
R-P1	같은 WARN 코드가 `notify.consecutive_failure_critical`(기본 3) 회 연속 실행에서 발생	CRITICAL 로 승격. 코드에 `_PERSIST` 접미사를 붙인 별개 알림 발생
R-P2	`runs` 에서 `status='FAILED'` 가 3일 연속	`CONSECUTIVE_FAILURES` CRITICAL 발생 (8절)
R-P3	WARN 토스트 표시가 3회 연속 실패(창 생성 예외)	해당 알림을 ERROR 로 승격해 모달 경로로 보냄
R-P4	서킷 브레이커가 `CLOSED -> OPEN` 으로 전환	전환 그 순간에만 CRITICAL 1회. OPEN 유지 동안은 침묵
R-P5	워치독이 heartbeat 나이 > `notify.watchdog_stale_minutes`(120) 판정	`WATCHDOG_STALE` CRITICAL

규칙	조건	결과
R-D1	다음 실행이 `SUCCESS` 로 끝남	일자성 WARN 코드 전부 `resolve(code, note="다음 실행 성공")`
R-D2	`checks.run_one(key)` 가 `ok=True` 로 바뀜	해당 상태성 CRITICAL 해소 (예: 재로그인 후 `AGY_AUTH`)
R-D3	서킷이 `OPEN -> HALF_OPEN -> CLOSED` 복귀	`*_CIRCUIT_OPEN` 해소 + INFO 1회("자동 복구됨")
R-D4	사용자가 모달에서 액션을 수행하고 성공 검증을 통과	즉시 해소. 모달을 닫기만 해서는 해소되지 않는다

해소는 사용자가 창을 닫는 것으로 이뤄지지 않는다. 반드시 checks 재실행 또는 다음 실행 성공이라는 객관적 증거가 있어야 한다. "닫으면 해결된 것으로 친다"는 조용한 실패의 교과서적 원인이다.

2.6 `alerts` / `alert_events` DDL

storage/migrations/0003_ops.sql 의 알림 관련 부분 전문이다.

-- ============================================================
-- 0003_ops.sql — 운영 계층 (알림 부분)
-- ============================================================

-- 발생 사실. 순수 append-only. 절대 UPDATE/DELETE 하지 않는다.
CREATE TABLE IF NOT EXISTS alert_events (
    event_id      INTEGER PRIMARY KEY AUTOINCREMENT,
    occurred_at   TEXT    NOT NULL,              -- ISO8601 with offset (+09:00)
    run_id        TEXT,                          -- 실행 밖(Agent 워치독)이면 NULL
    code          TEXT    NOT NULL,
    severity      TEXT    NOT NULL CHECK (severity IN ('INFO','WARN','ERROR','CRITICAL')),
    dedup_key     TEXT    NOT NULL,
    what          TEXT    NOT NULL,
    why           TEXT    NOT NULL,
    how           TEXT    NOT NULL,
    next_actions  TEXT    NOT NULL,              -- JSON 배열: ["run_now","open_log_dir"]
    context_json  TEXT    NOT NULL DEFAULT '{}', -- 템플릿 치환에 쓴 값 원본
    log_dir       TEXT,                          -- 절대경로
    source        TEXT    NOT NULL DEFAULT 'pipeline'  -- pipeline|watchdog|checks|agent
);
CREATE INDEX IF NOT EXISTS ix_alert_events_time ON alert_events(occurred_at DESC);
CREATE INDEX IF NOT EXISTS ix_alert_events_code ON alert_events(code, occurred_at DESC);

-- 표시·해소 상태 머신. dedup_key 당 1행.
CREATE TABLE IF NOT EXISTS alerts (
    alert_id        INTEGER PRIMARY KEY AUTOINCREMENT,
    dedup_key       TEXT    NOT NULL UNIQUE,
    code            TEXT    NOT NULL,
    severity        TEXT    NOT NULL CHECK (severity IN ('INFO','WARN','ERROR','CRITICAL')),
    first_seen_at   TEXT    NOT NULL,
    last_seen_at    TEXT    NOT NULL,
    occurrences     INTEGER NOT NULL DEFAULT 1,
    last_event_id   INTEGER NOT NULL REFERENCES alert_events(event_id),
    shown_at        TEXT,                        -- NULL 이면 표시 대기
    shown_channel   TEXT,                        -- toast|modal|messagebox|msgexe|webhook
    show_attempts   INTEGER NOT NULL DEFAULT 0,
    snoozed_until   TEXT,
    resolved_at     TEXT,
    resolved_note   TEXT
);
CREATE INDEX IF NOT EXISTS ix_alerts_pending
    ON alerts(resolved_at, shown_at, severity);
CREATE INDEX IF NOT EXISTS ix_alerts_code ON alerts(code);

-- 일일 표시 총량 상한(L3) 계산용 뷰
CREATE VIEW IF NOT EXISTS v_alert_shown_today AS
SELECT COUNT(*) AS shown_count
FROM   alerts
WHERE  shown_at IS NOT NULL
  AND  substr(shown_at, 1, 10) = strftime('%Y-%m-%d', 'now', 'localtime');

표시 대기 알림 조회 SQL (alerts.pending() 의 본체)

SELECT  a.alert_id, a.dedup_key, a.code, a.severity,
        a.first_seen_at, a.last_seen_at, a.occurrences, a.show_attempts,
        e.what, e.why, e.how, e.next_actions, e.context_json, e.log_dir, e.run_id
FROM    alerts a
JOIN    alert_events e ON e.event_id = a.last_event_id
WHERE   a.resolved_at IS NULL
  AND   (a.snoozed_until IS NULL OR a.snoozed_until < :now)
  AND   (
          a.shown_at IS NULL                                    -- 아직 못 보여줌
          OR (a.severity IN ('ERROR','CRITICAL')                -- 재촉 대상
              AND julianday(:now) - julianday(a.shown_at)
                  > :modal_repeat_minutes / 1440.0)
        )
ORDER BY CASE a.severity WHEN 'CRITICAL' THEN 0 WHEN 'ERROR' THEN 1
                         WHEN 'WARN'     THEN 2 ELSE 3 END,
         a.last_seen_at DESC;

연속 실패 일수 조회 SQL (8절 에스컬레이션용)

WITH daily AS (
    SELECT substr(started_at, 1, 10) AS d,
           MAX(CASE WHEN status IN ('SUCCESS','PARTIAL') THEN 1 ELSE 0 END) AS ok
    FROM   runs
    WHERE  started_at >= date('now', '-14 days')
    GROUP  BY d
),
ranked AS (
    SELECT d, ok, ROW_NUMBER() OVER (ORDER BY d DESC) AS rn
    FROM   daily
)
SELECT COALESCE(MIN(rn), 0) - 1 AS consecutive_failed_days
FROM   ranked
WHERE  ok = 1;
-- 성공 행이 하나도 없으면 결과가 -1 이 되므로 호출측에서
-- COUNT(*) FROM daily 로 대체 계산한다(코드 6.6 참조).

2.7 `state/alerts.json` 미러 스키마

Agent 프로세스가 SQLite 를 열지 않고도 알림 유무를 알 수 있어야 한다(배치가 DB 를 잠그고 있을 수 있고, PowerShell 폴백은 sqlite3 를 못 읽는다). 원자적 교체(os.replace)로 갱신한다.

{
  "schema": 1,
  "updated_at": "2026-09-02T06:04:11+09:00",
  "host": "DESKTOP-ABC",
  "pending": [
    {
      "alert_id": 42,
      "dedup_key": "AGY_AUTH",
      "code": "AGY_AUTH",
      "severity": "CRITICAL",
      "first_seen_at": "2026-09-01T06:03:55+09:00",
      "last_seen_at": "2026-09-02T06:03:58+09:00",
      "occurrences": 2,
      "title": "AI 요약을 만들지 못했습니다 — 로그인 만료",
      "what": "AI 요약 단계가 로그인 만료로 중단됐습니다.",
      "why": "Antigravity CLI(agy)의 Google 계정 인증이 만료되어 자동 갱신에 실패했습니다.",
      "how": "아래 [로그인 창 열기]를 누르고 Google 계정으로 다시 로그인하세요. 약 1분 걸립니다.",
      "next_actions": ["agy_relogin", "open_log_dir", "snooze", "dismiss"],
      "log_dir": "D:\\workspace\\DMF_Crawler\\logs\\run_20260902_060012",
      "run_id": "20260902_060012"
    }
  ],
  "counts": {"INFO": 0, "WARN": 1, "ERROR": 0, "CRITICAL": 1}
}

2.8 Windows 이벤트 로그 ID 배정

결정적 제약: eventcreate.exe 는 /ID 를 1~1000 범위로만 받는다. 범위를 벗어나면 오류: 잘못된 인수/옵션 으로 실패한다. 연구 문서 08 에 나온 1001 은 그대로 쓸 수 없다. 아래 표가 정본이다.

ID	`/T`	의미	기록 주체
100	INFORMATION	실행 시작	pipeline
110	SUCCESS	실행 성공(리포트 생성 완료)	pipeline
120	WARNING	실행 부분 성공(PARTIAL)	pipeline
130	INFORMATION	중복 실행 스킵(SKIPPED)	pipeline
200	WARNING	WARN 등급 알림 발생	alerts
300	ERROR	ERROR 등급 알림 발생(리포트 미생성)	alerts
400	ERROR	CRITICAL 등급 알림 발생	alerts
410	ERROR	워치독 — 배치 미실행 감지	watchdog
420	ERROR	연속 실패 임계 도달	watchdog
430	ERROR	자동 실행 일시중지(7일차 에스컬레이션)	watchdog
500	INFORMATION	알림 표시 성공	notify.pump
510	WARNING	알림 표시 실패 — 한 단계 강등(`NOTIFY_DEGRADED`)	notify.pump
520	WARNING	알림 표시 억제(일일 상한 도달)	notify.pump
900	ERROR	알리미 자신이 죽음(pump 예외)	notify.pump
910	ERROR	PowerShell 폴백 알림기가 발동함(= Python 계층이 죽었다는 증거)	notify.ps1

소스 이름: notify.eventlog_source = "DMF Crawler". eventcreate 의 /SO 는 이미 시스템에 등록된 원본 이름과 충돌하면 안 되고, 지정한 이름은 Application 로그에 자동 등록된다. 표준 사용자 권한으로 Application 로그 쓰기는 가능하다. (⚠️ 미검증: 그룹 정책으로 Application 로그 쓰기가 제한된 환경에서의 동작)

조회 명령

# 최근 20건
Get-WinEvent -FilterHashtable @{LogName='Application'; ProviderName='DMF Crawler'} -MaxEvents 20 |
    Format-Table TimeCreated, Id, LevelDisplayName, Message -AutoSize -Wrap

# CRITICAL 만
Get-WinEvent -FilterHashtable @{LogName='Application'; ProviderName='DMF Crawler'; Id=400,410,420,430} |
    Select-Object TimeCreated, Id, Message

3. 알림 문구 전집

3.1 4요소 계약

모든 알림은 아래 4요소를 반드시 갖는다. raise_alert() 는 하나라도 비면 ValueError 를 던진다 — 규격을 문서가 아니라 코드 계약으로 승격시킨다(아키텍처 §3.14).

요소	질문	작성 규칙
무엇 (`what`)	무엇이 실패했는가 / 언제	한 문장. 시각을 반드시 포함한다(`2026-09-02 06:04`). 기술 용어 대신 사용자가 아는 말로 — "fetch 스테이지"가 아니라 "식약처 데이터 받아오기"
왜 (`why`)	왜 그런가	한두 문장. 원인을 추정이 아니라 관측된 사실로 쓴다. 모르면 "원인을 특정하지 못했습니다"라고 정직하게 쓴다
어떻게 (`how`)	지금 무엇을 하면 되는가	명령형 한 문장 + 소요 시간. "약 1분 걸립니다" 같은 시간 표시가 착수율을 크게 올린다. 여러 방법이 있으면 가장 쉬운 것 하나만
다음 행동 (`next_actions`)	누를 것	액션 키 배열. 최소 1개(막다른 골목 금지, R7.7). `dismiss` 만 있는 실패 알림은 등록 거부된다

추가 필수 필드

필드	규칙
`title`	40자 이내. 토스트 제목줄에 잘리지 않아야 한다. 등급 접두어를 붙이지 않는다(색으로 표현)
`log_dir`	절대경로. 상대경로는 사용자가 어디서 여는지 몰라 쓸모없다
`where`	본문 말미에 자동 부착: `로그: <log_dir>`

금지 표현

금지	이유	대체
"오류가 발생했습니다"	아무 정보가 없다	실제로 무엇이 실패했는지
"관리자에게 문의하세요"	1인 운영에서 관리자는 본인이다	구체적 조치
예외 클래스명·스택트레이스	비개발자가 읽을 수 없다	사람의 말. 스택은 로그에만
"잠시 후 다시 시도하세요" (단독)	언제까지 기다릴지 모른다	"내일 06:00에 자동으로 다시 시도합니다"
API 키·토큰·URL 쿼리스트링	유출	앞 8자 지문만

3.2 시나리오 카탈로그

#	코드	등급	트리거	dedup 에 날짜 포함	쿨다운	기본 채널
S01	`RUN_OK`	INFO	실행 성공	예	1440분	로그·이벤트로그만
S02	`FETCH_FAILED`	WARN	재시도 4회 소진	예	240분	토스트
S03	`SOURCE_CIRCUIT_OPEN`	CRITICAL	서킷 CLOSED→OPEN 전환	아니오	60분	모달 + 웹훅
S04	`HTTP_BLOCKED_BODY`	WARN	200인데 본문이 빈/HTML/차단 안내	예	240분	토스트
S05	`ZERO_RECORDS`	CRITICAL	`totalCount`>0인데 수집 0건, 또는 `totalCount`=0	예	60분	모달 + 웹훅
S06	`INTEGRITY_BLOCKED`	WARN	무결성 게이트 2/4/5 차단	예	240분	토스트
S07	`INTEGRITY_DROP`	CRITICAL	게이트 3 — `totalCount` 5% 이상 급감	예	60분	모달 + 웹훅
S08	`SCHEMA_DRIFT`	CRITICAL	예상 필드 부재 / 널 비율 급증	예	60분	모달 + 웹훅
S09	`API_KEY_MISSING`	CRITICAL	DPAPI 복호화 결과 없음	아니오	60분	모달(강제)
S10	`API_KEY_INVALID`	CRITICAL	`resultCode != "00"` / HTTP 401·403	아니오	60분	모달(강제)
S11	`API_QUOTA_EXCEEDED`	WARN	일일 트래픽 초과 오류 코드	예	240분	토스트
S12	`AGY_MISSING`	CRITICAL	`agy.exe` 경로 부재	아니오	60분	모달
S13	`AGY_AUTH`	CRITICAL	`classify_error(env) == AUTH`	아니오	60분	모달 + 웹훅
S14	`AGY_QUOTA`	WARN	`classify_error(env) == QUOTA`	예	240분	토스트
S15	`AGY_TIMEOUT`	INFO	`--print-timeout` 초과	예	1440분	로그만
S16	`REPORT_LOCKED`	WARN	`os.replace` → `PermissionError` 3회	예	240분	토스트
S17	`REPORT_FAILED`	ERROR	report 스테이지 실패, 파일 없음	예	60분	모달
S18	`DISK_LOW`	WARN	여유 < `backup.min_free_gb`	예 + scope=드라이브	240분	토스트
S19	`DB_LOCKED`	ERROR	`sqlite3.OperationalError: database is locked`	예	60분	모달
S20	`DB_CORRUPT`	CRITICAL	`PRAGMA integrity_check` 실패	아니오	60분	모달(강제)
S21	`MIGRATION_FAILED`	CRITICAL	마이그레이션 중 예외	아니오	60분	모달(강제)
S22	`WATCHDOG_STALE`	CRITICAL	heartbeat 나이 > 120분	아니오	60분	모달 + 웹훅
S23	`TASK_MISSING`	CRITICAL	`checks` ⑨ 실패	아니오	60분	모달
S24	`CONSECUTIVE_FAILURES`	CRITICAL	3일 연속 FAILED	아니오	60분	모달 + 웹훅
S25	`PYTHON_BROKEN`	CRITICAL	pump 스탬프 미갱신(PowerShell 폴백이 판정)	아니오	240분	MessageBox
S26	`NOTIFY_DEGRADED`	WARN	표시 채널이 한 단계 떨어짐	예 + scope=채널	240분	이벤트로그만
S27	`RUN_PAUSED`	CRITICAL	7일 연속 실패로 자동 실행 중단	아니오	60분	모달 + 웹훅

"셀렉터 깨짐" 시나리오는 이 프로젝트에 존재하지 않는다. ADR-01/ADR-06 이 HTML 스크래핑과 브라우저 자동화를 코드에서 제거했으므로 CSS/XPath 셀렉터라는 개념 자체가 없다. 기능적으로 등가인 실패는 S08 SCHEMA_DRIFT(API 응답 필드 구조 변경) 이며, 아래 문구는 그 전제로 작성했다.

3.3 문구 전문

각 항목은 토스트(WARN) 또는 모달(ERROR/CRITICAL) 로 렌더링된 최종 결과다. {중괄호} 는 context_json 에서 치환된다.

S02 `FETCH_FAILED` — 크롤링(수집) 실패 · WARN

제목  식약처 데이터를 받아오지 못했습니다

본문  [무엇]  {run_date} {run_time} 실행에서 식약처 DMF 목록을 받아오지 못했습니다.
              {pages_ok}/{pages_total} 페이지까지만 받았습니다.
      [왜]    공공데이터포털 API 응답이 {attempts}회 연속 실패했습니다.
              마지막 오류: {last_error_short}
      [어떻게] 오늘 리포트는 마지막으로 성공한 {stale_date} 자료로 만들었습니다.
              내일 06:00에 자동으로 다시 시도합니다. 급하면 [지금 다시 실행]을 누르세요(약 2분).

      로그: {log_dir}

버튼  [지금 다시 실행]  [로그 열기]  [리포트 열기]  [닫기]

S03 `SOURCE_CIRCUIT_OPEN` — 소스 차단·연속 실패로 호출 중단 · CRITICAL

제목  식약처 API 호출을 24시간 중단했습니다

본문  [무엇]  {run_date} {run_time} 기준, 식약처 데이터 수집이 {fail_count}회 연속 실패해
              자동 호출을 중단(차단 회로 열림)했습니다.
      [왜]    같은 오류가 반복되면 상대 서버에 부담을 주고 차단당할 수 있어,
              {cooldown_hours}시간 동안 호출 자체를 멈춥니다.
              최근 오류: {last_error_short}
      [어떻게] 원인을 확인하려면 [진단 실행]을 누르세요(약 10초).
              원인이 해결됐다고 판단되면 [지금 다시 실행]으로 즉시 재시도할 수 있습니다.
              그대로 두면 {resume_at}에 자동으로 한 번 시험 호출합니다.

      로그: {log_dir}

버튼  [진단 실행]  [지금 다시 실행]  [로그 열기]  [나중에]

S04 `HTTP_BLOCKED_BODY` — 차단 감지(200인데 내용이 없음) · WARN

제목  API가 정상 응답 대신 안내 페이지를 보냈습니다

본문  [무엇]  {run_date} {run_time} 실행에서 {page_no}페이지 요청에 대해
              정상 코드(HTTP 200)가 왔지만 내용이 데이터가 아니었습니다.
      [왜]    응답 본문이 {body_bytes}바이트로 최소 기준({min_bytes}바이트)에 못 미치거나
              HTML 안내 페이지였습니다. 점검 중이거나 호출이 차단됐을 때 나타나는 형태입니다.
      [어떻게] 받은 원문을 그대로 보관했습니다. [원문 폴더 열기]에서 직접 확인할 수 있습니다.
              오늘 비교는 건너뛰었고 리포트는 이전 자료로 생성했습니다.
              내일 06:00에 자동 재시도합니다.

      원문: {raw_dir}
      로그: {log_dir}

버튼  [원문 폴더 열기]  [지금 다시 실행]  [로그 열기]  [닫기]

S05 `ZERO_RECORDS` — 0건 수집 · CRITICAL

제목  수집 결과가 0건입니다 — 비교를 중단했습니다

본문  [무엇]  {run_date} {run_time} 실행에서 원료의약품 등록(DMF) 자료를 한 건도 받지 못했습니다.
              (API가 알려준 전체 건수: {total_count}건, 실제 받은 건수: 0건)
      [왜]    0건을 그대로 받아들이면 어제 있던 {prev_count}건 전부가
              "취하됨"으로 잘못 기록됩니다. 그래서 저장과 비교를 모두 중단했습니다.
      [어떻게] 오늘 리포트는 {stale_date} 자료로 만들었고 맨 위에 경고 배너가 붙어 있습니다.
              [원문 폴더 열기]에서 실제 응답을 확인하세요(약 1분).
              공공데이터포털 점검 중이면 내일 자동 복구됩니다.

      원문: {raw_dir}
      로그: {log_dir}

버튼  [원문 폴더 열기]  [진단 실행]  [지금 다시 실행]  [나중에]

S06 `INTEGRITY_BLOCKED` — 무결성 게이트 차단 · WARN

제목  데이터 검증에 걸려 오늘 비교를 건너뛰었습니다

본문  [무엇]  {run_date} {run_time} 실행에서 안전장치 "{gate_name}"에 걸려
              오늘 자료를 기준값으로 저장하지 않았습니다.
      [왜]    {gate_detail}
              (기준: {gate_threshold} / 실측: {gate_observed})
              기준을 벗어난 자료를 저장하면 이후 모든 비교가 오염됩니다.
      [어떻게] 오늘 리포트는 정상 생성됐지만 "오늘 변경분" 시트는 비어 있습니다.
              받은 원문은 {raw_dir}에 보관돼 있습니다.
              내일 정상 수집되면 자동으로 복구됩니다. 확인이 필요하면 [원문 폴더 열기].

      로그: {log_dir}

버튼  [리포트 열기]  [원문 폴더 열기]  [로그 열기]  [닫기]

S07 `INTEGRITY_DROP` — 전체 건수 급감 · CRITICAL

제목  전체 등록 건수가 {drop_pct}% 줄었습니다 — 확인이 필요합니다

본문  [무엇]  {run_date} {run_time} 실행에서 전체 DMF 등록 건수가
              {prev_count}건 → {curr_count}건으로 {drop_count}건({drop_pct}%) 줄었습니다.
      [왜]    하루 만에 {threshold_pct}% 이상 줄어드는 것은 정상적인 변동이 아닙니다.
              대량 취하가 실제로 일어났거나, API가 일부 자료만 내려준 경우입니다.
              오탐으로 "전건 취하" 리포트가 나가는 것을 막기 위해 비교를 중단했습니다.
      [어떻게] [원문 폴더 열기]에서 오늘 받은 응답의 totalCount를 직접 확인하세요(약 2분).
              실제로 정상적인 감소라면 [지금 다시 실행 (검증 우회)]를 눌러 반영할 수 있습니다.

      원문: {raw_dir}
      로그: {log_dir}

버튼  [원문 폴더 열기]  [지금 다시 실행 (검증 우회)]  [로그 열기]  [나중에]

S08 `SCHEMA_DRIFT` — API 응답 구조 변경(구 "셀렉터 깨짐" 등가) · CRITICAL

제목  식약처 API 응답 형식이 바뀐 것 같습니다

본문  [무엇]  {run_date} {run_time} 실행에서 필수 항목 {missing_fields_str}을(를)
              응답에서 찾지 못했습니다. 값이 빈 비율이 {null_ratio_pct}%까지 올랐습니다.
              (평소 {baseline_null_pct}% 이하)
      [왜]    공공데이터포털 쪽에서 응답 필드 이름이나 구조를 바꿨을 가능성이 큽니다.
              잘못 해석한 자료를 저장하지 않기 위해 저장과 비교를 중단했습니다.
      [어떻게] 오늘 받은 원문 전체를 보관했습니다. 이 원문만 있으면 나중에 통째로
              다시 해석할 수 있으니 자료는 유실되지 않습니다.
              [원문 폴더 열기]로 실제 응답을 확인하고, 필드 이름이 바뀌었다면
              프로그램의 매핑을 고쳐야 합니다. AI 진단 결과가 있으면 {proposal_path}에 있습니다.

      원문: {raw_dir}
      로그: {log_dir}

버튼  [원문 폴더 열기]  [AI 진단 결과 열기]  [진단 실행]  [나중에]

AI 제안은 자동 적용되지 않는다(ADR-25). state/proposals/ 에 저장만 하고 사람이 승인한다. 이 버튼은 파일을 여는 것 이상을 하지 않는다.

S09 `API_KEY_MISSING` — API 키 미설정 · CRITICAL (강제 창)

제목  공공데이터포털 인증키가 없습니다

본문  [무엇]  {run_date} {run_time}에 배치를 시작하려 했으나
              공공데이터포털 인증키가 저장돼 있지 않아 실행하지 못했습니다.
      [왜]    이 프로그램은 식약처 원료의약품 등록 자료를 공공데이터포털 공식 API로 받아옵니다.
              API를 쓰려면 무료 인증키가 필요하고, 아직 한 번도 입력되지 않았습니다.
      [어떻게] [인증키 입력]을 누르면 입력 창이 열립니다.
              키가 없다면 [발급 페이지 열기]로 공공데이터포털에서 신청하세요(무료, 약 3분).
              입력하면 즉시 실제 호출로 검증한 뒤 이 PC에만 암호화해 저장합니다.

      * 이 알림은 키가 저장될 때까지 계속 표시됩니다.

버튼  [인증키 입력]  [발급 페이지 열기]  [나중에]

S10 `API_KEY_INVALID` — 인증키 무효·만료 · CRITICAL (강제 창)

제목  공공데이터포털 인증키가 거부됐습니다

본문  [무엇]  {run_date} {run_time} 실행에서 식약처 API가 인증키를 거부했습니다.
              (응답 코드: {result_code} / {result_msg})
      [왜]    저장된 키(앞 8자: {key_fingerprint}…)가 만료됐거나,
              해당 서비스의 활용 신청이 승인되지 않았거나, 키를 잘못 붙여넣었을 수 있습니다.
              특히 "인코딩 키"를 붙여넣으면 이 오류가 납니다 — **디코딩 키**를 써야 합니다.
      [어떻게] [인증키 다시 입력]을 눌러 공공데이터포털 마이페이지의
              **일반 인증키(Decoding)** 값을 붙여넣으세요(약 2분).
              입력 즉시 실제 호출로 검증하므로 맞는지 바로 알 수 있습니다.

      로그: {log_dir}

버튼  [인증키 다시 입력]  [마이페이지 열기]  [로그 열기]  [나중에]

S11 `API_QUOTA_EXCEEDED` — 일일 트래픽 초과 · WARN

제목  오늘 API 사용량을 모두 썼습니다

본문  [무엇]  {run_date} {run_time} 실행에서 공공데이터포털이
              일일 트래픽 초과를 알려왔습니다. ({error_code})
      [왜]    개발계정의 하루 호출 한도(기본 10,000회)를 넘었습니다.
              오늘 {calls_today}회 호출했습니다. 한도는 매일 자정에 초기화됩니다.
      [어떻게] 오늘은 더 호출하지 않고 마지막으로 성공한 {stale_date} 자료로 리포트를 만들었습니다.
              내일 06:00에 자동으로 정상 수집합니다. 아무것도 하지 않아도 됩니다.
              매일 반복된다면 공공데이터포털에서 운영계정 전환을 신청하세요.

      로그: {log_dir}

버튼  [리포트 열기]  [로그 열기]  [닫기]

S12 `AGY_MISSING` — agy 미설치 · CRITICAL

제목  AI 요약 도구(agy)가 설치돼 있지 않습니다

본문  [무엇]  {run_date} {run_time} 실행에서 AI 요약 단계를 건너뛰었습니다.
              Antigravity CLI(agy) 실행 파일을 찾지 못했습니다.
              찾아본 경로: {agy_path}
      [왜]    AI 요약은 매일 변경 내용을 자연어 브리핑으로 정리하는 부가 기능입니다.
              도구가 설치되지 않았거나 경로가 바뀌었습니다.
      [어떻게] [지금 설치]를 누르면 공식 설치 스크립트를 자동으로 실행합니다(약 2분, 인터넷 필요).
              설치 후 최초 1회 Google 계정 로그인이 필요하며 창이 자동으로 열립니다.
              AI 요약이 필요 없다면 [AI 기능 끄기]를 눌러 이 알림을 영구히 멈출 수 있습니다.

      * 오늘 리포트는 AI 요약 없이 정상 생성됐습니다.

버튼  [지금 설치]  [AI 기능 끄기]  [리포트 열기]  [나중에]

S13 `AGY_AUTH` — agy 인증 만료 · CRITICAL (강제 창)

제목  AI 요약을 만들지 못했습니다 — 로그인 만료

본문  [무엇]  {run_date} {run_time} 실행에서 AI 요약 단계가 중단됐습니다.
              {occurrences}일째 같은 상태입니다.
      [왜]    Antigravity CLI(agy)의 Google 계정 인증이 만료되어 자동 갱신에 실패했습니다.
              (agy 응답: {agy_error_short})
      [어떻게] [로그인 창 열기]를 누르면 검은 콘솔 창이 뜨고 브라우저가 열립니다.
              Google 계정으로 로그인한 뒤 창을 닫으면 끝입니다(약 1분).
              로그인이 끝나면 자동으로 확인해서 이 알림을 지웁니다.

      * 오늘 리포트는 AI 요약 없이 정상 생성됐습니다. 급하지 않다면 나중에 해도 됩니다.

      로그: {log_dir}

버튼  [로그인 창 열기]  [리포트 열기]  [로그 열기]  [나중에]

S14 `AGY_QUOTA` — agy 쿼터 소진 · WARN

제목  AI 요약을 건너뛰었습니다 — 사용 한도 소진

본문  [무엇]  {run_date} {run_time} 실행에서 AI 요약을 만들지 못했습니다.
      [왜]    Antigravity CLI의 모델 사용 한도를 다 썼습니다.
              (오늘 사용 토큰 {tokens_used}, 설정 상한 {tokens_cap})
      [어떻게] 아무것도 하지 않아도 됩니다. 내일 06:00에 자동으로 다시 시도합니다.
              오늘 리포트의 "대시보드" 시트 상단에 "AI 요약 없음: 사용 한도"라고 표시했습니다.
              자주 반복되면 config.toml 의 agy.daily_token_cap 을 조정하세요.

      로그: {log_dir}

버튼  [리포트 열기]  [로그 열기]  [닫기]

S16 `REPORT_LOCKED` — xlsx 파일 잠김 · WARN

제목  리포트를 다른 이름으로 저장했습니다

본문  [무엇]  {run_date} {run_time} 실행에서 오늘 리포트를 원래 파일 이름으로
              저장하지 못하고 "{fallback_name}"으로 저장했습니다.
      [왜]    "{target_name}" 파일이 Excel에서 열려 있어 덮어쓸 수 없었습니다.
              {retries}회 다시 시도했지만 계속 잠겨 있었습니다.
      [어떻게] Excel에서 해당 파일을 닫고 [리포트 다시 만들기]를 누르면
              원래 이름으로 정리됩니다(약 20초).
              지금 당장은 [폴더 열기]로 "{fallback_name}"을 그대로 열어 보셔도 됩니다.

      폴더: {report_dir}
      로그: {log_dir}

버튼  [리포트 다시 만들기]  [폴더 열기]  [닫기]

S17 `REPORT_FAILED` — 리포트 생성 실패 · ERROR (모달)

제목  오늘 리포트를 만들지 못했습니다

본문  [무엇]  {run_date} {run_time} 실행에서 Excel 리포트 생성 단계가 실패했습니다.
              오늘 날짜의 리포트 파일이 없습니다.
      [왜]    {failure_summary}
              (자료 수집과 비교는 정상적으로 끝났고 데이터베이스에 저장돼 있습니다.
               문제는 파일을 만드는 마지막 단계입니다.)
      [어떻게] 저장된 자료로 리포트만 다시 만들 수 있습니다.
              [리포트 다시 만들기]를 누르세요(약 20초, 인터넷 불필요).
              그래도 실패하면 [로그 열기]의 pipeline.log 마지막 30줄을 확인하세요.

      로그: {log_dir}

버튼  [리포트 다시 만들기]  [로그 열기]  [진단 실행]  [나중에]

S18 `DISK_LOW` — 디스크 부족 · WARN

제목  {drive} 드라이브 여유 공간이 부족합니다

본문  [무엇]  {run_date} {run_time} 실행에서 {drive} 드라이브 여유 공간이
              {free_gb}GB 남았습니다. (필요 최소 {min_gb}GB)
      [왜]    공간이 부족하면 데이터베이스 백업이 실패하고,
              더 줄어들면 리포트 저장도 실패합니다. 오늘은 백업만 건너뛰었습니다.
      [어떻게] [정리하기]를 누르면 보존 기간이 지난 로그·원문·오래된 백업을
              한 번에 지웁니다. 예상 확보량 약 {reclaim_mb}MB (약 10초).
              그래도 부족하면 config.toml 의 backup.dir 을 다른 드라이브로 바꾸세요.

      * 오늘 리포트는 정상 생성됐습니다.

      로그: {log_dir}

버튼  [정리하기]  [백업 폴더 열기]  [로그 열기]  [닫기]

S19 `DB_LOCKED` — 데이터베이스 잠김 · ERROR (모달)

제목  데이터베이스가 잠겨 있어 저장하지 못했습니다

본문  [무엇]  {run_date} {run_time} 실행에서 수집한 자료를 저장하지 못했습니다.
              {busy_timeout_s}초를 기다렸지만 데이터베이스가 계속 잠겨 있었습니다.
      [왜]    다른 프로그램이 dmf.sqlite3 파일을 붙잡고 있습니다.
              DB 브라우저 같은 도구를 열어 두었거나, 이전 실행이 아직 끝나지 않았을 수 있습니다.
      [어떻게] SQLite 관련 프로그램을 모두 닫고 [지금 다시 실행]을 누르세요(약 2분).
              아무것도 열어 둔 것이 없다면 PC를 재시작한 뒤 다시 시도하세요.

      DB: {db_path}
      로그: {log_dir}

버튼  [지금 다시 실행]  [DB 폴더 열기]  [로그 열기]  [나중에]

S20 `DB_CORRUPT` — 데이터베이스 손상 · CRITICAL (강제 창)

제목  데이터베이스 파일이 손상됐습니다

본문  [무엇]  {check_time} 점검에서 데이터베이스 무결성 검사가 실패했습니다.
              배치를 시작하지 않고 중단했습니다.
              (검사 결과: {integrity_result})
      [왜]    저장 중 강제 종료나 디스크 오류로 파일이 깨졌을 수 있습니다.
              손상된 파일에 계속 쓰면 남은 자료까지 잃습니다.
      [어떻게] 백업본이 {backup_count}개 있습니다. 가장 최근 것은 {latest_backup_date}입니다.
              [백업으로 복원]을 누르면 목록에서 고를 수 있습니다(약 30초).
              복원하면 그 날짜 이후 자료는 다시 수집해야 하며, 자동으로 채워집니다.

      현재 DB: {db_path}
      백업 폴더: {backup_dir}

버튼  [백업으로 복원]  [백업 폴더 열기]  [진단 실행]  [나중에]

S21 `MIGRATION_FAILED` — 스키마 마이그레이션 실패 · CRITICAL (강제 창)

제목  데이터베이스 업그레이드에 실패했습니다

본문  [무엇]  {check_time}에 데이터베이스 구조 업그레이드({from_version} → {to_version})가
              실패해 되돌렸습니다. 배치는 실행하지 않았습니다.
      [왜]    {failure_summary}
              변경은 트랜잭션으로 묶여 있어 **자료는 손상되지 않았습니다.**
      [어떻게] 업그레이드 직전 백업본을 아래 경로에 만들어 두었습니다.
              프로그램 버전을 이전으로 되돌리거나, 개발자에게 이 문구와 로그를 전달하세요.
              [백업 폴더 열기]로 백업본을 확인할 수 있습니다.

      업그레이드 직전 백업: {pre_migration_backup}
      로그: {log_dir}

버튼  [백업 폴더 열기]  [로그 열기]  [진단 실행]  [나중에]

S22 `WATCHDOG_STALE` — 배치 미실행 워치독 경보 · CRITICAL (강제 창)

제목  06:00 자동 실행이 되지 않았습니다

본문  [무엇]  지금 {now_time} 기준으로 오늘 06:00 배치가 실행된 흔적이 없습니다.
              마지막으로 성공한 실행은 {last_success_at} ({stale_hours}시간 전)입니다.
      [왜]    다음 중 하나입니다.
              · PC가 06:00에 꺼져 있었고 아직 따라잡기가 실행되지 않음
              · 작업 스케줄러 항목이 꺼졌거나 삭제됨
              · 실행이 시작됐지만 제한 시간({exec_limit_min}분)을 넘겨 강제 종료됨
      [어떻게] [지금 실행]을 누르면 즉시 오늘 자료를 수집합니다(약 2분).
              반복된다면 [작업 상태 확인]으로 스케줄러 등록 상태를 점검하세요.

      마지막 로그: {log_dir}

버튼  [지금 실행]  [작업 상태 확인]  [로그 폴더 열기]  [나중에]

S23 `TASK_MISSING` — 작업 스케줄러 등록 소실 · CRITICAL

제목  자동 실행 등록이 사라졌습니다

본문  [무엇]  {check_time} 점검에서 Windows 작업 스케줄러 항목 {missing_tasks_str}을(를)
              찾지 못했습니다.
      [왜]    시스템 정리 도구나 다른 프로그램이 지웠거나, 사용자가 직접 비활성화했을 수 있습니다.
              등록이 없으면 매일 06:00 자동 실행이 되지 않습니다.
      [어떻게] 의도적으로 끈 것이 아니라면 [자동 실행 다시 등록]을 누르세요(약 10초).
              같은 이름의 작업 3개를 다시 만듭니다. 기존 설정은 그대로 유지됩니다.

      * 자동으로 다시 등록하지 않는 이유: 사용자가 일부러 끈 것을 되살리면 안 되기 때문입니다.

버튼  [자동 실행 다시 등록]  [작업 스케줄러 열기]  [진단 실행]  [나중에]

S24 `CONSECUTIVE_FAILURES` — 연속 3일 실패 · CRITICAL (강제 창)

제목  {failed_days}일 연속으로 실패하고 있습니다

본문  [무엇]  {first_failed_date}부터 {last_failed_date}까지 {failed_days}일 연속
              배치가 실패했습니다. 그 기간 리포트가 만들어지지 않았습니다.
      [왜]    가장 많이 나온 원인 3가지입니다.
              1. {top_cause_1} ({top_cause_1_count}회)
              2. {top_cause_2} ({top_cause_2_count}회)
              3. {top_cause_3} ({top_cause_3_count}회)
      [어떻게] [전체 진단]을 누르면 전제조건 12가지를 한 번에 점검하고
              문제 항목마다 해결 버튼을 보여 줍니다(약 15초).
              대부분 인증키 또는 로그인 만료입니다.

      * {pause_day}일째에도 해결되지 않으면 자동 실행을 잠시 멈추고 다시 안내합니다.

      로그: {log_dir}

버튼  [전체 진단]  [지금 다시 실행]  [로그 폴더 열기]  [나중에]

S25 `PYTHON_BROKEN` — 알림 계층 자체가 죽음 · CRITICAL (MessageBox 폴백)

이 문구만은 PowerShell 이 렌더링한다(scripts/notify.ps1). Python 이 죽었을 때 뜨는 유일한 화면이므로 템플릿 치환을 최소화하고 상수 문자열 위주로 구성한다.

제목  DMF 크롤러 — 프로그램이 실행되지 않습니다

본문  [무엇]  DMF 크롤러의 알림 프로그램이 {stale_min}분째 응답하지 않습니다.
              (마지막 정상 동작: {last_pump_at})
      [왜]    Python 실행 환경(.venv)이 손상됐거나 삭제됐을 가능성이 큽니다.
              이 상태에서는 매일 06:00 자동 수집도 함께 멈춥니다.
      [어떻게] 프로젝트 폴더의 bootstrap.cmd 를 더블클릭해 다시 설치하세요(약 3분).
              기존 데이터와 설정은 그대로 유지됩니다.

      폴더: {project_root}
      Windows 이벤트 로그(Application) 원본 "DMF Crawler" 에 기록을 남겼습니다.

버튼  [확인]        ← MessageBox 는 버튼을 커스터마이즈하지 않는다(§6.1 참조)

S27 `RUN_PAUSED` — 자동 실행 일시중지 · CRITICAL (강제 창)

제목  자동 실행을 잠시 멈췄습니다

본문  [무엇]  {failed_days}일 연속 실패해 {pause_at}부터 매일 06:00 자동 실행을 멈췄습니다.
      [왜]    고쳐지지 않은 상태로 매일 같은 실패를 반복하면
              공공데이터포털에 불필요한 호출이 계속 쌓이기 때문입니다.
              멈춤은 작업 스케줄러를 지우는 것이 아니라 플래그 파일 하나로 이뤄집니다.
      [어떻게] [전체 진단]으로 원인을 먼저 해결한 뒤 [자동 실행 재개]를 누르세요.
              재개 버튼은 진단이 모두 통과해야 활성화됩니다.

      멈춤 플래그: {pause_flag_path}
      로그: {log_dir}

버튼  [전체 진단]  [자동 실행 재개]  [로그 폴더 열기]  [나중에]

나머지 INFO / 로그 전용 문구

코드	제목	본문(1줄 요약)
`RUN_OK`	오늘 리포트가 준비됐습니다	`{run_date} 수집 완료 — 신규 {new_n}건 / 변경 {chg_n}건 / 취하 {wdr_n}건. 리포트: {report_path}`
`AGY_TIMEOUT`	(표시 안 함)	`AI 요약이 {timeout}을 초과해 중단됨. 리포트는 정상 생성. 내일 자동 재시도.`
`NOTIFY_DEGRADED`	(표시 안 함)	`알림 표시가 {from_channel} → {to_channel} 로 강등됨. 사유: {reason}`
`NOTIFY_SUPPRESSED`	(표시 안 함)	`일일 알림 표시 상한 {cap}건 도달. 오늘 남은 알림 {pending_n}건은 표시하지 않음.`

3.4 `notify/messages.py` 전문

위 문구의 단일 정본이다. 문서와 코드가 어긋나지 않도록, 이 파일이 원본이고 §3.3 은 그 렌더링 결과다.

# src/dmf_crawler/notify/messages.py
"""알림 문구 템플릿 레지스트리.

계약(요구 R7.3):
    모든 템플릿은 what / why / how / next_actions 4요소를 갖는다.
    하나라도 비면 모듈 임포트 시점에 AssertionError 로 즉시 죽는다.
    -> 문구 누락이 06:00 런타임까지 잠복하지 않는다.
"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Mapping

from dmf_crawler.alerts import Severity

# ---------------------------------------------------------------- 액션 키
# gui/steps.py 의 ACTIONS 레지스트리와 키가 1:1로 일치해야 한다.
VALID_ACTIONS: frozenset[str] = frozenset({
    "run_now",            # 지금 다시 실행
    "run_now_force",      # 검증 우회 재실행 (--force)
    "report_only",        # 리포트만 다시 만들기
    "open_log_dir",       # 로그 폴더 열기
    "open_report_dir",    # 리포트 폴더 열기
    "open_report_file",   # 리포트 파일 바로 열기
    "open_raw_dir",       # API 원문 폴더 열기
    "open_backup_dir",    # 백업 폴더 열기
    "open_db_dir",        # DB 폴더 열기
    "open_proposal",      # AI 진단 결과 파일 열기
    "enter_api_key",      # 인증키 입력 창
    "open_api_portal",    # 공공데이터포털 발급/마이페이지 열기
    "agy_relogin",        # agy 재로그인 콘솔 열기
    "install_agy",        # agy 설치
    "disable_agy",        # AI 기능 끄기
    "reregister_tasks",   # 작업 스케줄러 3종 재등록
    "open_task_scheduler",# 작업 스케줄러 GUI 열기
    "restore_backup",     # 백업 복원 마법사
    "cleanup_disk",       # 로그/원문/백업 정리
    "open_doctor",        # 전체 진단 화면
    "resume_schedule",    # 자동 실행 재개(일시중지 해제)
    "snooze",             # 나중에 (기본 60분)
    "dismiss",            # 닫기
})

# 실패 알림에서 이것만 있으면 "막다른 골목"이다 (요구 R7.7).
_TERMINAL_ONLY = frozenset({"snooze", "dismiss"})


@dataclass(frozen=True, slots=True)
class AlertTemplate:
    code: str
    severity: Severity
    title: str
    what: str
    why: str
    how: str
    next_actions: tuple[str, ...]
    date_scoped: bool = True          # dedup_key 에 run_date 를 넣는가
    cooldown_minutes: int | None = None  # None 이면 등급 기본값 사용
    eventlog_id: int = 0              # 0 이면 등급 기본값 사용
    note: str = ""                    # 본문 하단 * 주석 (선택)

    def render(self, ctx: Mapping[str, Any]) -> "RenderedAlert":
        safe = _SafeDict(ctx)
        body_lines = [
            f"[무엇]   {self.what.format_map(safe)}",
            f"[왜]     {self.why.format_map(safe)}",
            f"[어떻게] {self.how.format_map(safe)}",
        ]
        if self.note:
            body_lines += ["", self.note.format_map(safe)]
        for label, key in (("원문", "raw_dir"), ("로그", "log_dir")):
            if ctx.get(key):
                body_lines += [f"{label}: {ctx[key]}"]
        return RenderedAlert(
            code=self.code,
            severity=self.severity,
            title=self.title.format_map(safe),
            what=self.what.format_map(safe),
            why=self.why.format_map(safe),
            how=self.how.format_map(safe),
            body="\n".join(body_lines),
            next_actions=self.next_actions,
        )


@dataclass(frozen=True, slots=True)
class RenderedAlert:
    code: str
    severity: Severity
    title: str
    what: str
    why: str
    how: str
    body: str
    next_actions: tuple[str, ...]


class _SafeDict(dict):
    """치환값이 없어도 죽지 않는다. 알림기는 절대 예외로 죽으면 안 된다."""

    def __init__(self, src: Mapping[str, Any]) -> None:
        super().__init__(src)

    def __missing__(self, key: str) -> str:  # noqa: D105
        return "(정보 없음)"


def _t(**kw: Any) -> AlertTemplate:
    return AlertTemplate(**kw)


TEMPLATES: dict[str, AlertTemplate] = {

    "RUN_OK": _t(
        code="RUN_OK", severity=Severity.INFO, eventlog_id=110,
        title="오늘 리포트가 준비됐습니다",
        what="{run_date} {run_time} 수집이 정상적으로 끝났습니다.",
        why="신규 {new_n}건 / 변경 {chg_n}건 / 취하 {wdr_n}건이 확인됐습니다.",
        how="리포트를 열어 확인하세요. 파일: {report_path}",
        next_actions=("open_report_file", "dismiss"),
        cooldown_minutes=1440,
    ),

    "FETCH_FAILED": _t(
        code="FETCH_FAILED", severity=Severity.WARN, eventlog_id=200,
        title="식약처 데이터를 받아오지 못했습니다",
        what="{run_date} {run_time} 실행에서 식약처 DMF 목록을 받아오지 못했습니다. "
             "{pages_ok}/{pages_total} 페이지까지만 받았습니다.",
        why="공공데이터포털 API 응답이 {attempts}회 연속 실패했습니다. "
            "마지막 오류: {last_error_short}",
        how="오늘 리포트는 마지막으로 성공한 {stale_date} 자료로 만들었습니다. "
            "내일 06:00에 자동으로 다시 시도합니다. "
            "급하면 [지금 다시 실행]을 누르세요(약 2분).",
        next_actions=("run_now", "open_log_dir", "open_report_file", "dismiss"),
    ),

    "SOURCE_CIRCUIT_OPEN": _t(
        code="SOURCE_CIRCUIT_OPEN", severity=Severity.CRITICAL, eventlog_id=400,
        date_scoped=False,
        title="식약처 API 호출을 {cooldown_hours}시간 중단했습니다",
        what="{run_date} {run_time} 기준, 식약처 데이터 수집이 {fail_count}회 연속 실패해 "
             "자동 호출을 중단(차단 회로 열림)했습니다.",
        why="같은 오류가 반복되면 상대 서버에 부담을 주고 차단당할 수 있어, "
            "{cooldown_hours}시간 동안 호출 자체를 멈춥니다. 최근 오류: {last_error_short}",
        how="원인을 확인하려면 [진단 실행]을 누르세요(약 10초). "
            "원인이 해결됐다고 판단되면 [지금 다시 실행]으로 즉시 재시도할 수 있습니다. "
            "그대로 두면 {resume_at}에 자동으로 한 번 시험 호출합니다.",
        next_actions=("open_doctor", "run_now", "open_log_dir", "snooze"),
    ),

    "HTTP_BLOCKED_BODY": _t(
        code="HTTP_BLOCKED_BODY", severity=Severity.WARN, eventlog_id=200,
        title="API가 정상 응답 대신 안내 페이지를 보냈습니다",
        what="{run_date} {run_time} 실행에서 {page_no}페이지 요청에 대해 "
             "정상 코드(HTTP 200)가 왔지만 내용이 데이터가 아니었습니다.",
        why="응답 본문이 {body_bytes}바이트로 최소 기준({min_bytes}바이트)에 못 미치거나 "
            "HTML 안내 페이지였습니다. 점검 중이거나 호출이 차단됐을 때 나타나는 형태입니다.",
        how="받은 원문을 그대로 보관했습니다. [원문 폴더 열기]에서 직접 확인할 수 있습니다. "
            "오늘 비교는 건너뛰었고 리포트는 이전 자료로 생성했습니다. "
            "내일 06:00에 자동 재시도합니다.",
        next_actions=("open_raw_dir", "run_now", "open_log_dir", "dismiss"),
    ),

    "ZERO_RECORDS": _t(
        code="ZERO_RECORDS", severity=Severity.CRITICAL, eventlog_id=400,
        title="수집 결과가 0건입니다 — 비교를 중단했습니다",
        what="{run_date} {run_time} 실행에서 원료의약품 등록(DMF) 자료를 한 건도 받지 못했습니다. "
             "(API가 알려준 전체 건수: {total_count}건, 실제 받은 건수: 0건)",
        why="0건을 그대로 받아들이면 어제 있던 {prev_count}건 전부가 '취하됨'으로 "
            "잘못 기록됩니다. 그래서 저장과 비교를 모두 중단했습니다.",
        how="오늘 리포트는 {stale_date} 자료로 만들었고 맨 위에 경고 배너가 붙어 있습니다. "
            "[원문 폴더 열기]에서 실제 응답을 확인하세요(약 1분). "
            "공공데이터포털 점검 중이면 내일 자동 복구됩니다.",
        next_actions=("open_raw_dir", "open_doctor", "run_now", "snooze"),
    ),

    "INTEGRITY_BLOCKED": _t(
        code="INTEGRITY_BLOCKED", severity=Severity.WARN, eventlog_id=200,
        title="데이터 검증에 걸려 오늘 비교를 건너뛰었습니다",
        what="{run_date} {run_time} 실행에서 안전장치 '{gate_name}'에 걸려 "
             "오늘 자료를 기준값으로 저장하지 않았습니다.",
        why="{gate_detail} (기준: {gate_threshold} / 실측: {gate_observed}) "
            "기준을 벗어난 자료를 저장하면 이후 모든 비교가 오염됩니다.",
        how="오늘 리포트는 정상 생성됐지만 '오늘 변경분' 시트는 비어 있습니다. "
            "받은 원문은 보관돼 있습니다. 내일 정상 수집되면 자동으로 복구됩니다. "
            "확인이 필요하면 [원문 폴더 열기]를 누르세요.",
        next_actions=("open_report_file", "open_raw_dir", "open_log_dir", "dismiss"),
    ),

    "INTEGRITY_DROP": _t(
        code="INTEGRITY_DROP", severity=Severity.CRITICAL, eventlog_id=400,
        title="전체 등록 건수가 {drop_pct}% 줄었습니다 — 확인이 필요합니다",
        what="{run_date} {run_time} 실행에서 전체 DMF 등록 건수가 "
             "{prev_count}건에서 {curr_count}건으로 {drop_count}건({drop_pct}%) 줄었습니다.",
        why="하루 만에 {threshold_pct}% 이상 줄어드는 것은 정상적인 변동이 아닙니다. "
            "대량 취하가 실제로 일어났거나, API가 일부 자료만 내려준 경우입니다. "
            "오탐으로 '전건 취하' 리포트가 나가는 것을 막기 위해 비교를 중단했습니다.",
        how="[원문 폴더 열기]에서 오늘 받은 응답의 totalCount를 직접 확인하세요(약 2분). "
            "실제로 정상적인 감소라면 [지금 다시 실행 (검증 우회)]를 눌러 반영할 수 있습니다.",
        next_actions=("open_raw_dir", "run_now_force", "open_log_dir", "snooze"),
    ),

    "SCHEMA_DRIFT": _t(
        code="SCHEMA_DRIFT", severity=Severity.CRITICAL, eventlog_id=400,
        title="식약처 API 응답 형식이 바뀐 것 같습니다",
        what="{run_date} {run_time} 실행에서 필수 항목 {missing_fields_str}을(를) "
             "응답에서 찾지 못했습니다. 값이 빈 비율이 {null_ratio_pct}%까지 올랐습니다. "
             "(평소 {baseline_null_pct}% 이하)",
        why="공공데이터포털 쪽에서 응답 필드 이름이나 구조를 바꿨을 가능성이 큽니다. "
            "잘못 해석한 자료를 저장하지 않기 위해 저장과 비교를 중단했습니다.",
        how="오늘 받은 원문 전체를 보관했습니다. 이 원문만 있으면 나중에 통째로 "
            "다시 해석할 수 있으니 자료는 유실되지 않습니다. "
            "[원문 폴더 열기]로 실제 응답을 확인하세요. "
            "AI 진단 결과가 있으면 [AI 진단 결과 열기]에서 볼 수 있습니다.",
        next_actions=("open_raw_dir", "open_proposal", "open_doctor", "snooze"),
        note="* AI가 제안한 수정은 자동으로 적용되지 않습니다. 사람이 확인한 뒤 반영합니다.",
    ),

    "API_KEY_MISSING": _t(
        code="API_KEY_MISSING", severity=Severity.CRITICAL, eventlog_id=400,
        date_scoped=False,
        title="공공데이터포털 인증키가 없습니다",
        what="{run_date} {run_time}에 배치를 시작하려 했으나 공공데이터포털 인증키가 "
             "저장돼 있지 않아 실행하지 못했습니다.",
        why="이 프로그램은 식약처 원료의약품 등록 자료를 공공데이터포털 공식 API로 받아옵니다. "
            "API를 쓰려면 무료 인증키가 필요하고, 아직 한 번도 입력되지 않았습니다.",
        how="[인증키 입력]을 누르면 입력 창이 열립니다. "
            "키가 없다면 [발급 페이지 열기]로 공공데이터포털에서 신청하세요(무료, 약 3분). "
            "입력하면 즉시 실제 호출로 검증한 뒤 이 PC에만 암호화해 저장합니다.",
        next_actions=("enter_api_key", "open_api_portal", "snooze"),
        note="* 이 알림은 키가 저장될 때까지 계속 표시됩니다.",
    ),

    "API_KEY_INVALID": _t(
        code="API_KEY_INVALID", severity=Severity.CRITICAL, eventlog_id=400,
        date_scoped=False,
        title="공공데이터포털 인증키가 거부됐습니다",
        what="{run_date} {run_time} 실행에서 식약처 API가 인증키를 거부했습니다. "
             "(응답 코드: {result_code} / {result_msg})",
        why="저장된 키(앞 8자: {key_fingerprint})가 만료됐거나, 해당 서비스의 활용 신청이 "
            "승인되지 않았거나, 키를 잘못 붙여넣었을 수 있습니다. "
            "특히 '인코딩 키'를 붙여넣으면 이 오류가 납니다 — 디코딩 키를 써야 합니다.",
        how="[인증키 다시 입력]을 눌러 공공데이터포털 마이페이지의 "
            "일반 인증키(Decoding) 값을 붙여넣으세요(약 2분). "
            "입력 즉시 실제 호출로 검증하므로 맞는지 바로 알 수 있습니다.",
        next_actions=("enter_api_key", "open_api_portal", "open_log_dir", "snooze"),
    ),

    "API_QUOTA_EXCEEDED": _t(
        code="API_QUOTA_EXCEEDED", severity=Severity.WARN, eventlog_id=200,
        title="오늘 API 사용량을 모두 썼습니다",
        what="{run_date} {run_time} 실행에서 공공데이터포털이 일일 트래픽 초과를 "
             "알려왔습니다. ({error_code})",
        why="개발계정의 하루 호출 한도(기본 10,000회)를 넘었습니다. "
            "오늘 {calls_today}회 호출했습니다. 한도는 매일 자정에 초기화됩니다.",
        how="오늘은 더 호출하지 않고 마지막으로 성공한 {stale_date} 자료로 리포트를 만들었습니다. "
            "내일 06:00에 자동으로 정상 수집합니다. 아무것도 하지 않아도 됩니다. "
            "매일 반복된다면 공공데이터포털에서 운영계정 전환을 신청하세요.",
        next_actions=("open_report_file", "open_log_dir", "dismiss"),
    ),

    "AGY_MISSING": _t(
        code="AGY_MISSING", severity=Severity.CRITICAL, eventlog_id=400,
        date_scoped=False,
        title="AI 요약 도구(agy)가 설치돼 있지 않습니다",
        what="{run_date} {run_time} 실행에서 AI 요약 단계를 건너뛰었습니다. "
             "Antigravity CLI(agy) 실행 파일을 찾지 못했습니다. 찾아본 경로: {agy_path}",
        why="AI 요약은 매일 변경 내용을 자연어 브리핑으로 정리하는 부가 기능입니다. "
            "도구가 설치되지 않았거나 경로가 바뀌었습니다.",
        how="[지금 설치]를 누르면 공식 설치 스크립트를 자동으로 실행합니다(약 2분, 인터넷 필요). "
            "설치 후 최초 1회 Google 계정 로그인이 필요하며 창이 자동으로 열립니다. "
            "AI 요약이 필요 없다면 [AI 기능 끄기]를 눌러 이 알림을 영구히 멈출 수 있습니다.",
        next_actions=("install_agy", "disable_agy", "open_report_file", "snooze"),
        note="* 오늘 리포트는 AI 요약 없이 정상 생성됐습니다.",
    ),

    "AGY_AUTH": _t(
        code="AGY_AUTH", severity=Severity.CRITICAL, eventlog_id=400,
        date_scoped=False,
        title="AI 요약을 만들지 못했습니다 — 로그인 만료",
        what="{run_date} {run_time} 실행에서 AI 요약 단계가 중단됐습니다. "
             "{occurrences}일째 같은 상태입니다.",
        why="Antigravity CLI(agy)의 Google 계정 인증이 만료되어 자동 갱신에 실패했습니다. "
            "(agy 응답: {agy_error_short})",
        how="[로그인 창 열기]를 누르면 검은 콘솔 창이 뜨고 브라우저가 열립니다. "
            "Google 계정으로 로그인한 뒤 창을 닫으면 끝입니다(약 1분). "
            "로그인이 끝나면 자동으로 확인해서 이 알림을 지웁니다.",
        next_actions=("agy_relogin", "open_report_file", "open_log_dir", "snooze"),
        note="* 오늘 리포트는 AI 요약 없이 정상 생성됐습니다. 급하지 않다면 나중에 해도 됩니다.",
    ),

    "AGY_QUOTA": _t(
        code="AGY_QUOTA", severity=Severity.WARN, eventlog_id=200,
        title="AI 요약을 건너뛰었습니다 — 사용 한도 소진",
        what="{run_date} {run_time} 실행에서 AI 요약을 만들지 못했습니다.",
        why="Antigravity CLI의 모델 사용 한도를 다 썼습니다. "
            "(오늘 사용 토큰 {tokens_used}, 설정 상한 {tokens_cap})",
        how="아무것도 하지 않아도 됩니다. 내일 06:00에 자동으로 다시 시도합니다. "
            "오늘 리포트의 '대시보드' 시트 상단에 'AI 요약 없음: 사용 한도'라고 표시했습니다. "
            "자주 반복되면 config.toml 의 agy.daily_token_cap 을 조정하세요.",
        next_actions=("open_report_file", "open_log_dir", "dismiss"),
    ),

    "AGY_TIMEOUT": _t(
        code="AGY_TIMEOUT", severity=Severity.INFO, eventlog_id=100,
        title="AI 요약이 시간 안에 끝나지 않았습니다",
        what="{run_date} {run_time} 실행에서 AI 요약이 {timeout} 안에 끝나지 않아 중단했습니다.",
        why="모델 응답이 느렸거나 네트워크가 불안정했습니다.",
        how="리포트는 정상 생성됐습니다. 내일 자동으로 다시 시도합니다.",
        next_actions=("dismiss",),
        cooldown_minutes=1440,
    ),

    "REPORT_LOCKED": _t(
        code="REPORT_LOCKED", severity=Severity.WARN, eventlog_id=200,
        title="리포트를 다른 이름으로 저장했습니다",
        what="{run_date} {run_time} 실행에서 오늘 리포트를 원래 파일 이름으로 저장하지 못하고 "
             "'{fallback_name}'으로 저장했습니다.",
        why="'{target_name}' 파일이 Excel에서 열려 있어 덮어쓸 수 없었습니다. "
            "{retries}회 다시 시도했지만 계속 잠겨 있었습니다.",
        how="Excel에서 해당 파일을 닫고 [리포트 다시 만들기]를 누르면 원래 이름으로 "
            "정리됩니다(약 20초). 지금 당장은 [폴더 열기]로 그대로 열어 보셔도 됩니다.",
        next_actions=("report_only", "open_report_dir", "dismiss"),
    ),

    "REPORT_FAILED": _t(
        code="REPORT_FAILED", severity=Severity.ERROR, eventlog_id=300,
        title="오늘 리포트를 만들지 못했습니다",
        what="{run_date} {run_time} 실행에서 Excel 리포트 생성 단계가 실패했습니다. "
             "오늘 날짜의 리포트 파일이 없습니다.",
        why="{failure_summary} 자료 수집과 비교는 정상적으로 끝났고 데이터베이스에 "
            "저장돼 있습니다. 문제는 파일을 만드는 마지막 단계입니다.",
        how="저장된 자료로 리포트만 다시 만들 수 있습니다. "
            "[리포트 다시 만들기]를 누르세요(약 20초, 인터넷 불필요). "
            "그래도 실패하면 [로그 열기]의 pipeline.log 마지막 30줄을 확인하세요.",
        next_actions=("report_only", "open_log_dir", "open_doctor", "snooze"),
    ),

    "DISK_LOW": _t(
        code="DISK_LOW", severity=Severity.WARN, eventlog_id=200,
        title="{drive} 드라이브 여유 공간이 부족합니다",
        what="{run_date} {run_time} 실행에서 {drive} 드라이브 여유 공간이 "
             "{free_gb}GB 남았습니다. (필요 최소 {min_gb}GB)",
        why="공간이 부족하면 데이터베이스 백업이 실패하고, 더 줄어들면 리포트 저장도 "
            "실패합니다. 오늘은 백업만 건너뛰었습니다.",
        how="[정리하기]를 누르면 보존 기간이 지난 로그·원문·오래된 백업을 한 번에 지웁니다. "
            "예상 확보량 약 {reclaim_mb}MB (약 10초). "
            "그래도 부족하면 config.toml 의 backup.dir 을 다른 드라이브로 바꾸세요.",
        next_actions=("cleanup_disk", "open_backup_dir", "open_log_dir", "dismiss"),
        note="* 오늘 리포트는 정상 생성됐습니다.",
    ),

    "DB_LOCKED": _t(
        code="DB_LOCKED", severity=Severity.ERROR, eventlog_id=300,
        title="데이터베이스가 잠겨 있어 저장하지 못했습니다",
        what="{run_date} {run_time} 실행에서 수집한 자료를 저장하지 못했습니다. "
             "{busy_timeout_s}초를 기다렸지만 데이터베이스가 계속 잠겨 있었습니다.",
        why="다른 프로그램이 dmf.sqlite3 파일을 붙잡고 있습니다. "
            "DB 브라우저 같은 도구를 열어 두었거나, 이전 실행이 아직 끝나지 않았을 수 있습니다.",
        how="SQLite 관련 프로그램을 모두 닫고 [지금 다시 실행]을 누르세요(약 2분). "
            "아무것도 열어 둔 것이 없다면 PC를 재시작한 뒤 다시 시도하세요.",
        next_actions=("run_now", "open_db_dir", "open_log_dir", "snooze"),
    ),

    "DB_CORRUPT": _t(
        code="DB_CORRUPT", severity=Severity.CRITICAL, eventlog_id=400,
        date_scoped=False,
        title="데이터베이스 파일이 손상됐습니다",
        what="{check_time} 점검에서 데이터베이스 무결성 검사가 실패했습니다. "
             "배치를 시작하지 않고 중단했습니다. (검사 결과: {integrity_result})",
        why="저장 중 강제 종료나 디스크 오류로 파일이 깨졌을 수 있습니다. "
            "손상된 파일에 계속 쓰면 남은 자료까지 잃습니다.",
        how="백업본이 {backup_count}개 있습니다. 가장 최근 것은 {latest_backup_date}입니다. "
            "[백업으로 복원]을 누르면 목록에서 고를 수 있습니다(약 30초). "
            "복원하면 그 날짜 이후 자료는 다시 수집해야 하며, 자동으로 채워집니다.",
        next_actions=("restore_backup", "open_backup_dir", "open_doctor", "snooze"),
    ),

    "MIGRATION_FAILED": _t(
        code="MIGRATION_FAILED", severity=Severity.CRITICAL, eventlog_id=400,
        date_scoped=False,
        title="데이터베이스 업그레이드에 실패했습니다",
        what="{check_time}에 데이터베이스 구조 업그레이드({from_version} → {to_version})가 "
             "실패해 되돌렸습니다. 배치는 실행하지 않았습니다.",
        why="{failure_summary} 변경은 트랜잭션으로 묶여 있어 자료는 손상되지 않았습니다.",
        how="업그레이드 직전 백업본을 {pre_migration_backup} 에 만들어 두었습니다. "
            "프로그램 버전을 이전으로 되돌리거나, 이 문구와 로그를 개발자에게 전달하세요.",
        next_actions=("open_backup_dir", "open_log_dir", "open_doctor", "snooze"),
    ),

    "WATCHDOG_STALE": _t(
        code="WATCHDOG_STALE", severity=Severity.CRITICAL, eventlog_id=410,
        date_scoped=False,
        title="06:00 자동 실행이 되지 않았습니다",
        what="지금 {now_time} 기준으로 오늘 06:00 배치가 실행된 흔적이 없습니다. "
             "마지막으로 성공한 실행은 {last_success_at} ({stale_hours}시간 전)입니다.",
        why="다음 중 하나입니다. "
            "(1) PC가 06:00에 꺼져 있었고 아직 따라잡기가 실행되지 않음 "
            "(2) 작업 스케줄러 항목이 꺼졌거나 삭제됨 "
            "(3) 실행이 시작됐지만 제한 시간({exec_limit_min}분)을 넘겨 강제 종료됨",
        how="[지금 실행]을 누르면 즉시 오늘 자료를 수집합니다(약 2분). "
            "반복된다면 [작업 상태 확인]으로 스케줄러 등록 상태를 점검하세요.",
        next_actions=("run_now", "open_task_scheduler", "open_log_dir", "snooze"),
    ),

    "TASK_MISSING": _t(
        code="TASK_MISSING", severity=Severity.CRITICAL, eventlog_id=400,
        date_scoped=False,
        title="자동 실행 등록이 사라졌습니다",
        what="{check_time} 점검에서 Windows 작업 스케줄러 항목 {missing_tasks_str}을(를) "
             "찾지 못했습니다.",
        why="시스템 정리 도구나 다른 프로그램이 지웠거나, 사용자가 직접 비활성화했을 수 "
            "있습니다. 등록이 없으면 매일 06:00 자동 실행이 되지 않습니다.",
        how="의도적으로 끈 것이 아니라면 [자동 실행 다시 등록]을 누르세요(약 10초). "
            "같은 이름의 작업 3개를 다시 만듭니다. 기존 설정은 그대로 유지됩니다.",
        next_actions=("reregister_tasks", "open_task_scheduler", "open_doctor", "snooze"),
        note="* 자동으로 다시 등록하지 않는 이유: 사용자가 일부러 끈 것을 되살리면 안 되기 때문입니다.",
    ),

    "CONSECUTIVE_FAILURES": _t(
        code="CONSECUTIVE_FAILURES", severity=Severity.CRITICAL, eventlog_id=420,
        date_scoped=False,
        title="{failed_days}일 연속으로 실패하고 있습니다",
        what="{first_failed_date}부터 {last_failed_date}까지 {failed_days}일 연속 배치가 "
             "실패했습니다. 그 기간 리포트가 만들어지지 않았습니다.",
        why="가장 많이 나온 원인 3가지입니다. "
            "(1) {top_cause_1} ({top_cause_1_count}회) "
            "(2) {top_cause_2} ({top_cause_2_count}회) "
            "(3) {top_cause_3} ({top_cause_3_count}회)",
        how="[전체 진단]을 누르면 전제조건 12가지를 한 번에 점검하고 "
            "문제 항목마다 해결 버튼을 보여 줍니다(약 15초). "
            "대부분 인증키 또는 로그인 만료입니다.",
        next_actions=("open_doctor", "run_now", "open_log_dir", "snooze"),
        note="* {pause_day}일째에도 해결되지 않으면 자동 실행을 잠시 멈추고 다시 안내합니다.",
    ),

    "RUN_PAUSED": _t(
        code="RUN_PAUSED", severity=Severity.CRITICAL, eventlog_id=430,
        date_scoped=False,
        title="자동 실행을 잠시 멈췄습니다",
        what="{failed_days}일 연속 실패해 {pause_at}부터 매일 06:00 자동 실행을 멈췄습니다.",
        why="고쳐지지 않은 상태로 매일 같은 실패를 반복하면 공공데이터포털에 "
            "불필요한 호출이 계속 쌓이기 때문입니다. "
            "멈춤은 작업 스케줄러를 지우는 것이 아니라 플래그 파일 하나로 이뤄집니다.",
        how="[전체 진단]으로 원인을 먼저 해결한 뒤 [자동 실행 재개]를 누르세요. "
            "재개 버튼은 진단이 모두 통과해야 활성화됩니다.",
        next_actions=("open_doctor", "resume_schedule", "open_log_dir", "snooze"),
    ),

    "NOTIFY_DEGRADED": _t(
        code="NOTIFY_DEGRADED", severity=Severity.WARN, eventlog_id=510,
        title="알림 표시 방식이 바뀌었습니다",
        what="{now_time}에 알림을 {from_channel} 방식으로 띄우지 못했습니다.",
        why="{reason}",
        how="{to_channel} 방식으로 대신 표시했습니다. 반복되면 [진단 실행]을 눌러 확인하세요.",
        next_actions=("open_doctor", "dismiss"),
    ),

    "NOTIFY_SUPPRESSED": _t(
        code="NOTIFY_SUPPRESSED", severity=Severity.INFO, eventlog_id=520,
        title="오늘 알림 표시를 멈췄습니다",
        what="{now_time} 기준 오늘 알림 표시가 상한 {cap}건에 도달했습니다.",
        why="같은 문제가 반복돼 알림이 지나치게 많이 뜨는 것을 막기 위한 조치입니다.",
        how="남은 {pending_n}건은 [전체 진단] 화면에서 한꺼번에 확인할 수 있습니다.",
        next_actions=("open_doctor", "dismiss"),
        cooldown_minutes=1440,
    ),
}


# ---------------------------------------------------- 임포트 시점 계약 검증
def _validate_registry() -> None:
    for code, tpl in TEMPLATES.items():
        assert tpl.code == code, f"{code}: code 필드 불일치"
        for field_name in ("title", "what", "why", "how"):
            value = getattr(tpl, field_name)
            assert value and value.strip(), f"{code}: {field_name} 가 비었다 (요구 R7.3)"
        assert tpl.next_actions, f"{code}: next_actions 가 비었다 (요구 R7.7)"
        unknown = set(tpl.next_actions) - VALID_ACTIONS
        assert not unknown, f"{code}: 알 수 없는 액션 키 {sorted(unknown)}"
        if tpl.severity.at_least(Severity.WARN):
            actionable = set(tpl.next_actions) - _TERMINAL_ONLY
            assert actionable, (
                f"{code}: 실패 알림에 실행 가능한 액션이 없다 — 막다른 골목 금지(R7.7)"
            )
        assert len(tpl.title) <= 60, f"{code}: title 이 너무 길다 ({len(tpl.title)}자)"


_validate_registry()


def render(code: str, ctx: Mapping[str, Any]) -> RenderedAlert:
    """알림 코드와 컨텍스트로 최종 문구를 만든다. 미등록 코드도 죽지 않는다."""
    tpl = TEMPLATES.get(code)
    if tpl is None:
        return RenderedAlert(
            code=code, severity=Severity.ERROR,
            title=f"알 수 없는 오류가 기록됐습니다 ({code})",
            what=f"{code} 오류가 발생했습니다.",
            why="이 오류에 대한 안내 문구가 아직 등록되지 않았습니다.",
            how="[로그 열기]로 pipeline.log 를 확인하세요.",
            body=(f"[무엇]   {code} 오류가 발생했습니다.\n"
                  f"[왜]     이 오류에 대한 안내 문구가 아직 등록되지 않았습니다.\n"
                  f"[어떻게] [로그 열기]로 pipeline.log 를 확인하세요.\n"
                  f"로그: {ctx.get('log_dir', '(정보 없음)')}"),
            next_actions=("open_log_dir", "open_doctor", "dismiss"),
        )
    return tpl.render(ctx)

4. 버튼 액션 구현

4.1 액션 키 레지스트리

토스트 버튼과 복구 GUI 버튼은 같은 레지스트리를 호출한다. 액션은 src/dmf_crawler/gui/steps.py 에 산다(아키텍처 트리: "단계별 액션: 키 입력·발급페이지 열기·agy 설치·재로그인·작업 등록").

액션 키	라벨	하는 일	창을 닫는가	성공 검증
`run_now`	지금 다시 실행	`dmf run --trigger manual` 을 자식 프로세스로 기동, 진행 창 표시	아니오(진행률로 전환)	종료 코드 0
`run_now_force`	지금 다시 실행 (검증 우회)	`dmf run --trigger manual --force`	아니오	종료 코드 0
`report_only`	리포트 다시 만들기	`dmf report-only`	아니오	리포트 파일 mtime 갱신
`open_log_dir`	로그 열기	`explorer.exe <log_dir>`	아니오	—
`open_report_dir`	폴더 열기	`explorer.exe <report_dir>`	아니오	—
`open_report_file`	리포트 열기	`os.startfile(report_path)`	예	—
`open_raw_dir`	원문 폴더 열기	`explorer.exe <data/raw/YYYY-MM-DD>`	아니오	—
`open_backup_dir`	백업 폴더 열기	`explorer.exe <backup_dir>`	아니오	—
`open_db_dir`	DB 폴더 열기	`explorer.exe /select,<db_path>`	아니오	—
`open_proposal`	AI 진단 결과 열기	`os.startfile(proposal_path)`, 없으면 폴더	아니오	—
`enter_api_key`	인증키 입력	키 입력 다이얼로그 → 실호출 검증 → DPAPI 저장	아니오	`checks.run_one("api_key_live")`
`open_api_portal`	발급 페이지 열기	`webbrowser.open(공공데이터포털 URL)`	아니오	—
`agy_relogin`	로그인 창 열기	새 콘솔 창에 대화형 `agy` 기동 (§5)	아니오	`checks.run_one("agy_auth")`
`install_agy`	지금 설치	`scripts/bootstrap_agy.ps1` 무인 실행 + 진행률	아니오	`agy --version` 성공
`disable_agy`	AI 기능 끄기	`config.local.toml` 에 `[agy] enabled = false` 기록	예	설정 재로드
`reregister_tasks`	자동 실행 다시 등록	`dmf install-task`	아니오	`checks.run_one("tasks")`
`open_task_scheduler`	작업 스케줄러 열기	`mmc.exe taskschd.msc`	아니오	—
`restore_backup`	백업으로 복원	백업 목록 다이얼로그 → 현재 DB 대피 → 복사	아니오	`PRAGMA integrity_check`
`cleanup_disk`	정리하기	보존 기간 지난 로그·원문·백업 삭제 후 확보량 표시	아니오	여유 공간 재측정
`open_doctor`	전체 진단 / 진단 실행	`gui.app.launch(mode="inspect")`	아니오	—
`resume_schedule`	자동 실행 재개	`state/paused.flag` 삭제. 진단 전부 통과해야 활성화	예	플래그 부재
`snooze`	나중에	`alerts.snooze(alert_id, minutes)`	예	—
`dismiss`	닫기	`alerts.mark_shown(alert_id)` 만 하고 닫음	예	—

설계 규칙 3개

모든 액션은 예외를 던지지 않는다. ActionResult(ok, message) 를 반환한다. 액션이 죽어서 복구 창이 사라지는 것은 최악이다.
성공 검증이 있는 액션은 검증을 통과해야만 알림을 해소한다(R-D4). 버튼을 눌렀다는 사실만으로는 해소하지 않는다.
모든 액션은 logs/agent/actions.jsonl 에 기록된다. 사용자가 무엇을 눌렀고 결과가 무엇이었는지 사후 추적이 가능해야 "고쳤는데 또 뜬다"를 진단할 수 있다.

4.2 `gui/steps.py` 전문

# src/dmf_crawler/gui/steps.py
"""알림·복구 화면의 버튼 액션 구현.

계약:
    - 모든 액션은 ActionResult 를 반환하며 예외를 밖으로 던지지 않는다.
    - 액션 키는 notify.messages.VALID_ACTIONS 와 1:1 로 일치한다.
    - 액션 실행은 전부 logs/agent/actions.jsonl 에 기록된다.
"""
from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
import webbrowser
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Callable, Mapping

from dmf_crawler import paths
from dmf_crawler.config import Config

# 공공데이터포털 — 인증키 발급/조회
API_PORTAL_URL = "https://www.data.go.kr/iim/api/selectAPIAcountView.do"
# agy 공식 설치 스크립트 (docs/research/05a-agy-cli-ssot.md §3.2)
AGY_INSTALL_URL = "https://antigravity.google/cli/install.ps1"

# 콘솔 창을 만들지 않는다. pythonw 로 떠 있는 GUI 에서 검은 창이 번쩍이면 안 된다.
CREATE_NO_WINDOW = 0x08000000
# 새 콘솔 창을 "보이게" 만든다. agy 재로그인에서만 쓴다.
CREATE_NEW_CONSOLE = 0x00000010


@dataclass(frozen=True, slots=True)
class ActionResult:
    ok: bool
    message: str
    close_window: bool = False
    verify_check_key: str | None = None   # 성공 검증에 쓸 checks 키


# ------------------------------------------------------------------ 유틸
def _venv_python(windowed: bool = False) -> Path:
    """현재 프로젝트 .venv 의 인터프리터. windowed=True 면 pythonw."""
    exe = "pythonw.exe" if windowed else "python.exe"
    candidate = paths.PROJECT_ROOT / ".venv" / "Scripts" / exe
    if candidate.exists():
        return candidate
    return Path(sys.executable)


def _explorer(target: Path, select: bool = False) -> ActionResult:
    if not target.exists():
        parent = target.parent
        if not parent.exists():
            return ActionResult(False, f"경로를 찾을 수 없습니다: {target}")
        target, select = parent, False
    args = ["explorer.exe"]
    args.append(f"/select,{target}" if select else str(target))
    # explorer.exe 는 성공해도 종료 코드 1 을 반환하는 일이 잦다. 코드를 보지 않는다.
    subprocess.Popen(args, creationflags=CREATE_NO_WINDOW)
    return ActionResult(True, f"탐색기를 열었습니다: {target}")


def _spawn_cli(cfg: Config, argv: list[str], *, new_console: bool = False,
               env_extra: Mapping[str, str] | None = None) -> subprocess.Popen:
    env = os.environ.copy()
    # 배치 중 agy 자동 업데이트가 끼어드는 것을 막는다 (agy SSOT §3.4)
    env["AGY_CLI_DISABLE_AUTO_UPDATE"] = "true"
    if env_extra:
        env.update(env_extra)
    flags = CREATE_NEW_CONSOLE if new_console else CREATE_NO_WINDOW
    py = _venv_python(windowed=not new_console)
    return subprocess.Popen(
        [str(py), "-m", "dmf_crawler", *argv],
        cwd=str(paths.PROJECT_ROOT),
        env=env,
        creationflags=flags,
    )


def _log_action(key: str, result: ActionResult, ctx: Mapping[str, Any]) -> None:
    """액션 감사 로그. 실패해도 조용히 넘어간다."""
    try:
        log_dir = paths.LOGS_DIR / "agent"
        log_dir.mkdir(parents=True, exist_ok=True)
        record = {
            "at": datetime.now().astimezone().isoformat(timespec="seconds"),
            "action": key,
            "ok": result.ok,
            "message": result.message,
            "alert_code": ctx.get("code"),
            "alert_id": ctx.get("alert_id"),
        }
        with (log_dir / "actions.jsonl").open("a", encoding="utf-8") as fh:
            fh.write(json.dumps(record, ensure_ascii=False) + "\n")
    except Exception:
        pass


# ------------------------------------------------------------ 액션 구현부
def act_run_now(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    _spawn_cli(cfg, ["run", "--trigger", "manual"])
    return ActionResult(True, "수집을 시작했습니다. 약 2분 걸립니다.",
                        verify_check_key="last_run")


def act_run_now_force(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    _spawn_cli(cfg, ["run", "--trigger", "manual", "--force"])
    return ActionResult(True, "검증을 우회해 수집을 시작했습니다.",
                        verify_check_key="last_run")


def act_report_only(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    _spawn_cli(cfg, ["report-only"])
    return ActionResult(True, "리포트를 다시 만들고 있습니다. 약 20초 걸립니다.",
                        verify_check_key="report_writable")


def act_open_log_dir(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    raw = ctx.get("log_dir")
    target = Path(raw) if raw else paths.LOGS_DIR
    return _explorer(target)


def act_open_report_dir(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    return _explorer(paths.reports_dir(cfg))


def act_open_report_file(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    raw = ctx.get("report_path")
    target = Path(raw) if raw else paths.latest_report_path(cfg)
    if not target.exists():
        return _explorer(paths.reports_dir(cfg))
    try:
        os.startfile(str(target))  # noqa: S606 — Windows 전용, 의도된 호출
    except OSError as exc:
        return ActionResult(False, f"리포트를 열지 못했습니다: {exc}")
    return ActionResult(True, "리포트를 열었습니다.", close_window=True)


def act_open_raw_dir(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    raw = ctx.get("raw_dir")
    target = Path(raw) if raw else paths.RAW_DIR
    return _explorer(target)


def act_open_backup_dir(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    return _explorer(paths.backup_dir(cfg))


def act_open_db_dir(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    return _explorer(paths.db_path(cfg), select=True)


def act_open_proposal(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    raw = ctx.get("proposal_path")
    if raw and Path(raw).exists():
        try:
            os.startfile(raw)  # noqa: S606
            return ActionResult(True, "AI 진단 결과를 열었습니다.")
        except OSError as exc:
            return ActionResult(False, f"파일을 열지 못했습니다: {exc}")
    return _explorer(paths.PROPOSALS_DIR)


def act_open_api_portal(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    webbrowser.open(API_PORTAL_URL)
    return ActionResult(True, "브라우저에서 공공데이터포털을 열었습니다.")


def act_open_task_scheduler(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    subprocess.Popen(["mmc.exe", "taskschd.msc"], creationflags=CREATE_NO_WINDOW)
    return ActionResult(True, "작업 스케줄러를 열었습니다. "
                              "'작업 스케줄러 라이브러리'에서 DMF_Crawler 로 시작하는 "
                              "항목 3개를 확인하세요.")


def act_reregister_tasks(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    proc = _spawn_cli(cfg, ["install-task"])
    try:
        rc = proc.wait(timeout=60)
    except subprocess.TimeoutExpired:
        return ActionResult(False, "등록이 60초 안에 끝나지 않았습니다. "
                                   "관리자 권한이 필요할 수 있습니다.")
    if rc != 0:
        return ActionResult(False, f"등록에 실패했습니다(코드 {rc}). "
                                   "이 창을 관리자 권한으로 다시 실행해 보세요.")
    return ActionResult(True, "자동 실행 작업 3개를 다시 등록했습니다.",
                        verify_check_key="tasks")


def act_disable_agy(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    local = paths.CONFIG_DIR / "config.local.toml"
    try:
        existing = local.read_text(encoding="utf-8") if local.exists() else ""
        if "[agy]" in existing:
            lines = []
            in_agy = False
            wrote = False
            for line in existing.splitlines():
                stripped = line.strip()
                if stripped.startswith("["):
                    if in_agy and not wrote:
                        lines.append("enabled = false")
                        wrote = True
                    in_agy = stripped == "[agy]"
                if in_agy and stripped.startswith("enabled"):
                    lines.append("enabled = false")
                    wrote = True
                    continue
                lines.append(line)
            if in_agy and not wrote:
                lines.append("enabled = false")
            new_text = "\n".join(lines) + "\n"
        else:
            new_text = existing.rstrip() + "\n\n[agy]\nenabled = false\n"
        tmp = local.with_suffix(".toml.tmp")
        tmp.write_text(new_text, encoding="utf-8")
        os.replace(tmp, local)
    except OSError as exc:
        return ActionResult(False, f"설정을 저장하지 못했습니다: {exc}")
    return ActionResult(True, "AI 요약 기능을 껐습니다. 리포트는 계속 생성됩니다.",
                        close_window=True)


def act_cleanup_disk(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    freed = 0
    now = datetime.now()

    def _prune(root: Path, days: int) -> None:
        nonlocal freed
        if not root.exists():
            return
        cutoff = now - timedelta(days=days)
        for child in sorted(root.iterdir()):
            try:
                mtime = datetime.fromtimestamp(child.stat().st_mtime)
                if mtime >= cutoff:
                    continue
                size = sum(f.stat().st_size for f in child.rglob("*") if f.is_file()) \
                    if child.is_dir() else child.stat().st_size
                if child.is_dir():
                    shutil.rmtree(child, ignore_errors=True)
                else:
                    child.unlink(missing_ok=True)
                freed += size
            except OSError:
                continue

    _prune(paths.LOGS_DIR, cfg.logging.retain_days)
    _prune(paths.RAW_DIR, cfg.source.archive_retain_days)

    # 백업은 날짜가 아니라 개수 기준(보존 정책과 동일)
    backups = sorted(paths.backup_dir(cfg).glob("dmf_*.sqlite3"))
    for old in backups[:-cfg.backup.keep_count] if len(backups) > cfg.backup.keep_count else []:
        try:
            freed += old.stat().st_size
            old.unlink()
        except OSError:
            continue

    mb = freed / (1024 * 1024)
    return ActionResult(True, f"{mb:,.0f}MB 를 정리했습니다.",
                        verify_check_key="disk_free")


def act_snooze(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    minutes = int(ctx.get("snooze_minutes", cfg.notify.snooze_minutes))
    return ActionResult(True, f"{minutes}분 뒤에 다시 알려드립니다.", close_window=True)


def act_dismiss(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    return ActionResult(True, "닫았습니다.", close_window=True)


# --- 아래 4개는 별도 절에서 상세히 다룬다 -------------------------------
#     act_enter_api_key   : 온보딩 마법사 명세(docs/design/04-onboarding-wizard.md)
#     act_restore_backup  : 백업 복원 마법사(같은 문서)
#     act_open_doctor     : gui.app.launch(mode="inspect")
#     act_install_agy     : §5.4
#     act_agy_relogin     : §5.2  <- 이 문서의 핵심
def act_open_doctor(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    from dmf_crawler.gui import app as gui_app
    gui_app.launch(mode="inspect")
    return ActionResult(True, "진단 화면을 열었습니다.")


def act_resume_schedule(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    from dmf_crawler import checks
    failing = [c for c in checks.run_all(cfg) if not c.ok]
    if failing:
        names = ", ".join(c.title for c in failing[:3])
        return ActionResult(False,
                            f"아직 해결되지 않은 항목이 있습니다: {names}. "
                            f"먼저 [전체 진단]에서 해결하세요.")
    try:
        paths.PAUSE_FLAG.unlink(missing_ok=True)
    except OSError as exc:
        return ActionResult(False, f"멈춤 해제에 실패했습니다: {exc}")
    return ActionResult(True, "자동 실행을 재개했습니다. 내일 06:00부터 정상 동작합니다.",
                        close_window=True)


# ------------------------------------------------------------- 레지스트리
ACTIONS: dict[str, tuple[str, Callable[[Config, Mapping[str, Any]], ActionResult]]] = {
    "run_now":             ("지금 다시 실행",            act_run_now),
    "run_now_force":       ("지금 다시 실행 (검증 우회)", act_run_now_force),
    "report_only":         ("리포트 다시 만들기",        act_report_only),
    "open_log_dir":        ("로그 열기",                 act_open_log_dir),
    "open_report_dir":     ("폴더 열기",                 act_open_report_dir),
    "open_report_file":    ("리포트 열기",               act_open_report_file),
    "open_raw_dir":        ("원문 폴더 열기",            act_open_raw_dir),
    "open_backup_dir":     ("백업 폴더 열기",            act_open_backup_dir),
    "open_db_dir":         ("DB 폴더 열기",              act_open_db_dir),
    "open_proposal":       ("AI 진단 결과 열기",         act_open_proposal),
    "open_api_portal":     ("발급 페이지 열기",          act_open_api_portal),
    "open_task_scheduler": ("작업 스케줄러 열기",        act_open_task_scheduler),
    "reregister_tasks":    ("자동 실행 다시 등록",       act_reregister_tasks),
    "disable_agy":         ("AI 기능 끄기",              act_disable_agy),
    "cleanup_disk":        ("정리하기",                  act_cleanup_disk),
    "open_doctor":         ("전체 진단",                 act_open_doctor),
    "resume_schedule":     ("자동 실행 재개",            act_resume_schedule),
    "snooze":              ("나중에",                    act_snooze),
    "dismiss":             ("닫기",                      act_dismiss),
    # §5 에서 정의
    "agy_relogin":         ("로그인 창 열기",            None),   # noqa: 아래에서 채움
    "install_agy":         ("지금 설치",                 None),
    # 온보딩 마법사 문서에서 정의
    "enter_api_key":       ("인증키 입력",               None),
    "restore_backup":      ("백업으로 복원",             None),
}


def label_of(key: str) -> str:
    entry = ACTIONS.get(key)
    return entry[0] if entry else key


def invoke(key: str, cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    """액션 실행 단일 진입점. 절대 예외를 던지지 않는다."""
    entry = ACTIONS.get(key)
    if entry is None or entry[1] is None:
        result = ActionResult(False, f"'{key}' 동작이 아직 구현되지 않았습니다.")
        _log_action(key, result, ctx)
        return result
    try:
        result = entry[1](cfg, ctx)
    except Exception as exc:  # noqa: BLE001 — 액션은 절대 창을 죽이면 안 된다
        result = ActionResult(False, f"동작 중 문제가 생겼습니다: {exc}")
    _log_action(key, result, ctx)
    return result

4.3 성공 검증과 알림 해소의 연결

버튼을 누른 뒤 검증 키가 있는 액션은 다음 흐름을 탄다.

# gui/app.py 안 (발췌)
def on_action_clicked(self, alert: AlertRow, action_key: str) -> None:
    result = steps.invoke(action_key, self.cfg, alert.as_context())
    self.status_bar.set(result.message, ok=result.ok)

    if result.ok and result.verify_check_key:
        # 즉시 검증하지 않고, 최대 120초 동안 2초 간격으로 폴링한다.
        # (run_now 처럼 자식 프로세스가 끝나야 결과가 나오는 액션이 있다)
        self.after(2000, lambda: self._poll_verify(alert, result.verify_check_key, 0))

    if result.close_window:
        if action_key == "snooze":
            alerts.snooze(self.conn, alert.alert_id,
                          minutes=self.cfg.notify.snooze_minutes)
        else:
            alerts.mark_shown(self.conn, alert.alert_id, channel="modal")
        self.destroy()


def _poll_verify(self, alert: AlertRow, check_key: str, elapsed_s: int) -> None:
    outcome = checks.run_one(check_key, self.cfg)
    if outcome.ok:
        alerts.resolve(self.conn, alert.code,
                       note=f"사용자 조치 후 {check_key} 통과")
        self.status_bar.set("해결됐습니다. 이 알림을 닫습니다.", ok=True)
        self.after(1500, self.destroy)
        return
    if elapsed_s >= 120:
        self.status_bar.set(
            f"아직 해결되지 않았습니다: {outcome.detail}", ok=False)
        return
    self.after(2000, lambda: self._poll_verify(alert, check_key, elapsed_s + 2))

왜 폴링인가: run_now 는 자식 프로세스를 띄우고 즉시 반환한다. 버튼을 누른 직후 검증하면 항상 실패한다. 사용자가 "고쳤는데 왜 안 없어지지"라고 느끼는 지점이 정확히 여기다. 최대 120초 폴링이 이 간극을 메운다.

4.4 프로토콜 핸들러 (`dmf://`) — 지금은 불필요, 등록 방법은 완비

결론부터: 기본 경로에서는 등록하지 않는다. 우리 토스트는 tkinter 창이고, 버튼은 같은 프로세스 안에서 steps.invoke() 를 직접 호출한다. OS 를 경유할 이유가 없다.

필요해지는 경우는 두 가지다.

진짜 Windows 토스트(액션 센터 잔류)를 선택 의존성으로 켜는 날. XML 토스트의 <action activationType="protocol" arguments="dmf://run_now"> 는 프로토콜 등록 없이는 동작하지 않는다.
리포트 xlsx 안에서 하이퍼링크로 액션을 걸고 싶을 때(예: 대시보드 시트의 "지금 다시 실행" 링크).

등록 방법 — HKCU 만 쓴다(관리자 권한 불필요).

# scripts/register_protocol.ps1
# dmf:// 프로토콜 핸들러를 현재 사용자에게만 등록한다. 관리자 권한이 필요 없다.
[CmdletBinding()]
param(
    [string]$ProjectRoot = (Split-Path -Parent $PSScriptRoot),
    [switch]$Unregister
)

$ErrorActionPreference = 'Stop'
$key = 'HKCU:\Software\Classes\dmf'

if ($Unregister) {
    if (Test-Path $key) { Remove-Item $key -Recurse -Force }
    Write-Host 'dmf:// 프로토콜 등록을 해제했습니다.'
    return
}

$pythonw = Join-Path $ProjectRoot '.venv\Scripts\pythonw.exe'
if (-not (Test-Path $pythonw)) { throw "pythonw.exe 를 찾을 수 없습니다: $pythonw" }

# "URL Protocol" 이라는 (값이 빈) 값 이름이 있어야 셸이 프로토콜로 인식한다.
New-Item -Path $key -Force | Out-Null
Set-ItemProperty -Path $key -Name '(default)'   -Value 'URL:DMF Crawler Protocol'
Set-ItemProperty -Path $key -Name 'URL Protocol' -Value ''

New-Item -Path "$key\DefaultIcon" -Force | Out-Null
Set-ItemProperty -Path "$key\DefaultIcon" -Name '(default)' -Value "$pythonw,0"

New-Item -Path "$key\shell\open\command" -Force | Out-Null
# %1 은 전체 URL(dmf://run_now?alert_id=42)이 통째로 넘어온다. 반드시 큰따옴표로 감싼다.
$cmd = "`"$pythonw`" -m dmf_crawler handle-uri `"%1`""
Set-ItemProperty -Path "$key\shell\open\command" -Name '(default)' -Value $cmd

Write-Host "dmf:// 프로토콜을 등록했습니다."
Write-Host "테스트: Win+R 에 다음을 입력하세요 →  dmf://open_doctor"

동등한 .reg 파일 (수동 배포용)

Windows Registry Editor Version 5.00

[HKEY_CURRENT_USER\Software\Classes\dmf]
@="URL:DMF Crawler Protocol"
"URL Protocol"=""

[HKEY_CURRENT_USER\Software\Classes\dmf\DefaultIcon]
@="D:\\workspace\\DMF_Crawler\\.venv\\Scripts\\pythonw.exe,0"

[HKEY_CURRENT_USER\Software\Classes\dmf\shell\open\command]
@="\"D:\\workspace\\DMF_Crawler\\.venv\\Scripts\\pythonw.exe\" -m dmf_crawler handle-uri \"%1\""

수신측 — dmf handle-uri 서브커맨드

URI 는 외부에서 들어오는 신뢰할 수 없는 입력이다. 화이트리스트 검증 없이 실행하면 임의 명령 실행 취약점이 된다.

# src/dmf_crawler/cli.py (발췌) — handle-uri 구현
from urllib.parse import urlparse, parse_qs

def cmd_handle_uri(args) -> int:
    """dmf://<action_key>?alert_id=<n> 형태만 허용한다."""
    from dmf_crawler.gui import steps
    from dmf_crawler.config import load_config
    from dmf_crawler.notify import messages

    parsed = urlparse(args.uri)
    if parsed.scheme != "dmf":
        return 2
    # netloc 에 액션 키가 온다: dmf://run_now  ->  netloc == "run_now"
    action_key = (parsed.netloc or parsed.path.lstrip("/")).strip().lower()

    # 화이트리스트 검증. 등록되지 않은 키는 무조건 거부한다.
    if action_key not in messages.VALID_ACTIONS:
        return 2

    qs = parse_qs(parsed.query)
    ctx: dict[str, object] = {}
    raw_id = qs.get("alert_id", [""])[0]
    if raw_id.isdigit():                       # 숫자만 허용
        ctx["alert_id"] = int(raw_id)

    cfg = load_config()
    result = steps.invoke(action_key, cfg, ctx)
    return 0 if result.ok else 1

검증 항목	규칙
scheme	`dmf` 가 아니면 즉시 거부
액션 키	`VALID_ACTIONS` 화이트리스트에 없으면 거부. 경로·명령 문자열을 URI 에서 받지 않는다
`alert_id`	숫자만. 그 외 쿼리 파라미터는 전부 버린다
파일 경로	URI 로 절대 받지 않는다. 경로는 항상 `paths.py` 와 DB 에서만 온다

5. agy 재로그인 유도

5.1 문제 정의와 Session 0 제약

사실	출처	함의
`agy` OAuth 토큰은 `~/.gemini/antigravity-cli/antigravity-oauth-token` 평문 파일	agy SSOT §4.2 실측	작업 스케줄러를 동일 사용자 계정으로 돌리면 인증이 통과한다. SYSTEM 계정 금지
Windows Credential Manager 에는 항목이 없다	agy SSOT §4.2 실측 (`cmdkey /list` 무결과)	S4U(암호 미저장)로 실행해도 키링 잠금 문제가 없다. 다만 사용자 프로필이 로드돼야 한다
`access_token` 은 만료된다	agy SSOT §4.2	refresh 실패 시 배치가 인증 오류로 죽는다. 미인증 감지 → 사용자 알림 경로가 필수
최초 1회는 대화형 로그인이 필수	agy SSOT §0	헤드리스는 캐시된 자격증명만 쓴다. 로그인은 브라우저 OAuth 를 요구한다
로그인 흐름은 기본 브라우저를 연다	agy SSOT §4.1	데스크톱이 없는 세션에서는 브라우저가 뜨지 않는다

Session 0 제약이 만드는 막힘

06:00 DMF_Crawler_Daily (S4U, 비대화형)
   -> agy -p ... 실행
   -> 토큰 만료, refresh 실패
   -> agy 가 브라우저를 열려고 시도
   -> S4U 세션에는 데스크톱이 없다 -> 아무 창도 안 뜬다
   -> agy 가 stderr 로 인증 프롬프트를 출력하지만 읽을 사람이 없다
   -> --print-timeout 까지 대기하다 타임아웃
   ===> 배치가 침묵 속에 매일 실패한다

우회 = 구조적 분리 (ADR-10/11)

[배치 S4U]  agy 실패 -> classify_error == AUTH
              -> AgyEnvelope 를 값으로 반환 (예외 없음)
              -> alerts.raise_alert(code="AGY_AUTH", severity=CRITICAL)
              -> 리포트는 AI 요약 없이 정상 생성, 종료 코드 0
                    |
                    | (최대 15분)
                    v
[Agent Interactive]  pending() 에서 AGY_AUTH 발견
              -> 강제 모달 표시, [로그인 창 열기] 버튼
                    |
                    | 사용자 클릭
                    v
[새 콘솔 창]  CREATE_NEW_CONSOLE 로 agy 대화형 기동
              -> agy 가 기본 브라우저를 연다 (여기는 데스크톱이 있다)
              -> 사용자가 Google 로그인
              -> 토큰 파일 갱신
                    |
                    v
[Agent]      토큰 파일 mtime 변화 감지 -> 헬스 프롬프트 1회로 실증
              -> 성공하면 alerts.resolve("AGY_AUTH")

절대 하지 말 것

금지	이유
배치(S4U)에서 `CREATE_NEW_CONSOLE` 로 agy 를 띄우기	Session 0 에 콘솔이 만들어지고 아무도 못 본다. 프로세스만 영원히 남는다
`psexec -i 1` 등으로 세션 주입	관리자 권한·보안 소프트웨어 충돌. 로그온 세션이 없으면 여전히 실패
토큰 파일을 코드가 직접 갱신	리프레시 프로토콜을 재구현하는 것. agy 가 바뀌면 즉시 깨진다
`GEMINI_API_KEY` 로 전환해 로그인 자체를 없애기	가능은 하다(agy SSOT §4.1). 그러나 다른 과금 체계로 넘어가는 결정이므로 알림 설계가 임의로 할 수 없다. 부록에 미결로 남긴다

5.2 `act_agy_relogin` 전문

# src/dmf_crawler/gui/steps.py (이어서)
"""agy 재로그인 — 새 콘솔 창을 띄우는 유일한 액션."""
from __future__ import annotations

import subprocess
import threading
import time
from pathlib import Path
from typing import Any, Mapping

AGY_TOKEN_PATH = Path.home() / ".gemini" / "antigravity-cli" / "antigravity-oauth-token"
AGY_DEFAULT_EXE = Path(os.environ.get("LOCALAPPDATA", "")) / "agy" / "bin" / "agy.exe"

# 사용자가 콘솔 창에서 무엇을 해야 하는지 안내하는 배너.
# agy 자체는 한국어 안내를 하지 않으므로 우리가 감싼다.
_RELOGIN_BANNER = r"""
@echo off
chcp 65001 > nul
title DMF 크롤러 - Antigravity CLI 로그인
echo.
echo ============================================================
echo   Antigravity CLI (agy) 로그인
echo ============================================================
echo.
echo   1. 잠시 후 기본 브라우저가 자동으로 열립니다.
echo   2. Google 계정으로 로그인하세요.
echo   3. 브라우저에 "로그인 완료" 가 뜨면 이 창으로 돌아오세요.
echo   4. 이 창에 프롬프트가 보이면  /quit  를 입력하고 Enter 를 누르세요.
echo.
echo   * 브라우저가 열리지 않으면 이 창에 표시되는 URL 을 복사해
echo     브라우저 주소창에 붙여넣으세요.
echo.
echo ============================================================
echo.
"%AGY_EXE%"
echo.
echo ============================================================
echo   로그인 절차가 끝났습니다. 이 창은 닫아도 됩니다.
echo   DMF 크롤러 창으로 돌아가면 자동으로 확인합니다.
echo ============================================================
echo.
pause
"""


def _agy_exe(cfg: Config) -> Path:
    configured = (cfg.agy.binary_path or "").strip()
    if configured:
        return Path(configured)
    return AGY_DEFAULT_EXE


def act_agy_relogin(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    """새 콘솔 창에서 대화형 agy 를 띄운다.

    핵심:
      - CREATE_NEW_CONSOLE  : 반드시 보이는 콘솔이어야 한다. pythonw 에서 뜨는
                              자식은 부모의 콘솔을 물려받지 않으므로 이 플래그가 없으면
                              stdin 이 없어 agy 가 즉시 종료된다.
      - AGY_CLI_DISABLE_AUTO_UPDATE : 로그인 중 바이너리가 교체되는 사고를 막는다.
      - .cmd 래퍼          : agy 를 직접 띄우면 사용자가 무엇을 해야 하는지 모른다.
                              한국어 안내 배너를 앞뒤로 감싼다.
    """
    exe = _agy_exe(cfg)
    if not exe.exists():
        return ActionResult(
            False,
            f"agy 실행 파일이 없습니다: {exe}\n"
            f"먼저 [지금 설치]를 눌러 설치하세요.")

    # 래퍼 배치 파일을 로그 디렉터리에 만든다(임시 폴더는 백신이 막는 경우가 있다).
    wrapper_dir = paths.STATE_DIR / "tmp"
    wrapper_dir.mkdir(parents=True, exist_ok=True)
    wrapper = wrapper_dir / "agy_login.cmd"
    try:
        wrapper.write_text(_RELOGIN_BANNER, encoding="utf-8")
    except OSError as exc:
        return ActionResult(False, f"로그인 도우미 파일을 만들지 못했습니다: {exc}")

    env = os.environ.copy()
    env["AGY_EXE"] = str(exe)
    env["AGY_CLI_DISABLE_AUTO_UPDATE"] = "true"

    token_mtime_before = _token_mtime()

    try:
        subprocess.Popen(
            ["cmd.exe", "/c", str(wrapper)],
            cwd=str(paths.PROJECT_ROOT),
            env=env,
            creationflags=CREATE_NEW_CONSOLE,   # <- 이것이 전부다
            close_fds=True,
        )
    except OSError as exc:
        return ActionResult(False, f"로그인 창을 열지 못했습니다: {exc}")

    # 백그라운드로 토큰 파일 변화를 지켜본다. 최대 10분.
    threading.Thread(
        target=_watch_token_change,
        args=(cfg, token_mtime_before),
        daemon=True,
        name="agy-token-watch",
    ).start()

    return ActionResult(
        True,
        "로그인 창을 열었습니다. 브라우저에서 Google 계정으로 로그인하세요. "
        "완료되면 이 창이 자동으로 확인합니다.",
        verify_check_key="agy_auth")


def _token_mtime() -> float:
    try:
        return AGY_TOKEN_PATH.stat().st_mtime
    except OSError:
        return 0.0


def _watch_token_change(cfg: Config, before: float, timeout_s: int = 600) -> None:
    """토큰 파일이 갱신되면 실제 호출 1회로 인증을 실증한다."""
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        time.sleep(3)
        if _token_mtime() <= before:
            continue
        # 파일이 갱신됐다. 쓰기가 끝날 시간을 준 뒤 실증한다.
        time.sleep(2)
        if verify_agy_auth(cfg).ok:
            return
        # 갱신은 됐는데 실증 실패 -> 계속 지켜본다(사용자가 재시도 중일 수 있다)
        before = _token_mtime()


def verify_agy_auth(cfg: Config) -> ActionResult:
    """헤드리스 최소 호출로 인증 상태를 실증한다.

    agy SSOT §4.3 의 권장 헬스체크를 그대로 옮긴 것이다.
    주의: 이 호출도 토큰을 소비한다. 재로그인 직후와 checks 화면에서만 호출하고,
          06:00 배치에서는 절대 호출하지 않는다(ADR-16: 인증 상태는 실작업 결과로 판정).
    """
    exe = _agy_exe(cfg)
    if not exe.exists():
        return ActionResult(False, "agy 실행 파일이 없습니다.")

    env = os.environ.copy()
    env["AGY_CLI_DISABLE_AUTO_UPDATE"] = "true"
    try:
        proc = subprocess.run(
            [str(exe), "-p", "Reply with exactly: PONG",
             "--output-format", "json", "--print-timeout", "90s"],
            capture_output=True, text=True, encoding="utf-8", errors="replace",
            env=env, timeout=120, creationflags=CREATE_NO_WINDOW,
        )
    except subprocess.TimeoutExpired:
        return ActionResult(False, "확인 요청이 시간 안에 끝나지 않았습니다.")
    except OSError as exc:
        return ActionResult(False, f"agy 를 실행하지 못했습니다: {exc}")

    # agy SSOT §13: 종료 코드와 status 를 둘 다 확인해야 한다.
    if proc.returncode != 0:
        return ActionResult(False, f"agy 가 오류로 끝났습니다(코드 {proc.returncode}).")

    from dmf_crawler.agy import extract
    envelope = extract.parse_envelope(proc.stdout)
    if envelope is None:
        return ActionResult(False, "agy 응답을 해석하지 못했습니다.")
    if envelope.get("status") != "SUCCESS":
        return ActionResult(False, f"인증이 아직 유효하지 않습니다: {envelope.get('error')}")
    return ActionResult(True, "agy 로그인이 정상 확인됐습니다.")

CREATE_NEW_CONSOLE 이 반드시 필요한 이유

실행 주체	콘솔 상속	`agy` 대화형 동작
`pythonw.exe`(GUI) 에서 플래그 없이 `Popen`	부모에 콘솔이 없으므로 자식도 없음	stdin 부재 → 즉시 종료. 사용자는 아무것도 못 본다
`CREATE_NO_WINDOW`	콘솔은 생기지만 숨겨짐	프롬프트가 보이지 않아 로그인 코드를 붙여넣을 수 없다
`CREATE_NEW_CONSOLE`	새 콘솔이 보이게 생성	✅ 정답. 브라우저가 열리고 프롬프트가 보인다
`start` 를 셸로 호출(`shell=True`)	동작은 하지만 인자 이스케이프가 취약	경로에 공백이 있으면 깨진다. 쓰지 않는다

5.3 재로그인 성공 판정

세 단계로 판정한다. 하나라도 건너뛰면 오탐이 난다.

단계	판정	실패 시
1. 토큰 파일 mtime 변화	`antigravity-oauth-token` 의 mtime 이 클릭 시점보다 커졌는가	계속 대기(최대 10분)
2. 토큰 파일 형태 확인	JSON 이고 `token.access_token` 키가 있는가. 값은 절대 로그에 남기지 않는다	손상 판정 → 재로그인 재안내
3. 실호출 실증	`agy -p "Reply with exactly: PONG"` 가 종료 코드 0 + `status == "SUCCESS"`	만료 상태 유지 → 알림 해소하지 않음

3단계를 반드시 하는 이유: 파일이 갱신됐다고 인증이 유효하다는 보장이 없다. 다른 계정으로 로그인했거나, 쓰기가 중간에 끊겼거나, 조직 정책으로 토큰이 즉시 무효화됐을 수 있다. "파일이 바뀌었으니 됐겠지"는 조용한 실패의 전형이다.

단, 3단계는 토큰을 소비한다. 그래서 호출 시점을 엄격히 제한한다.

호출해도 되는 곳	호출하면 안 되는 곳
재로그인 직후(토큰 mtime 변화 감지 시)	06:00 배치의 preflight
`dmf doctor` / 진단 GUI 를 사용자가 직접 열었을 때	Agent 의 15분 주기 pump
온보딩 마법사의 agy 단계	`checks.run_all()` 의 자동 실행 경로

checks ⑦ "agy 인증 상태"는 기본적으로 agy_calls 테이블의 최근 결과로 판정한다(아키텍처 §3.15). 실호출은 사용자가 명시적으로 요청했을 때만 한다.

5.4 `act_install_agy` 전문

# src/dmf_crawler/gui/steps.py (이어서)

def act_install_agy(cfg: Config, ctx: Mapping[str, Any]) -> ActionResult:
    """scripts/bootstrap_agy.ps1 을 무인 실행한다.

    설치가 끝나면 최초 1회 로그인이 필요하므로, 성공 시 곧바로
    act_agy_relogin 을 이어서 호출한다.
    """
    script = paths.SCRIPTS_DIR / "bootstrap_agy.ps1"
    if not script.exists():
        return ActionResult(False, f"설치 스크립트를 찾을 수 없습니다: {script}")

    try:
        proc = subprocess.run(
            ["powershell.exe", "-NoProfile", "-NonInteractive",
             "-ExecutionPolicy", "Bypass", "-File", str(script)],
            capture_output=True, text=True, encoding="utf-8", errors="replace",
            timeout=300, creationflags=CREATE_NO_WINDOW,
        )
    except subprocess.TimeoutExpired:
        return ActionResult(False, "설치가 5분 안에 끝나지 않았습니다. "
                                   "인터넷 연결을 확인하고 다시 시도하세요.")
    except OSError as exc:
        return ActionResult(False, f"설치 스크립트를 실행하지 못했습니다: {exc}")

    if proc.returncode != 0:
        tail = (proc.stderr or proc.stdout or "").strip().splitlines()[-3:]
        return ActionResult(False, "설치에 실패했습니다.\n" + "\n".join(tail))

    if not _agy_exe(cfg).exists():
        return ActionResult(False, "설치는 끝났지만 실행 파일을 찾지 못했습니다. "
                                   "PC를 재시작한 뒤 다시 시도하세요.")

    # 설치 직후에는 반드시 로그인이 필요하다. 바로 이어서 띄운다.
    relogin = act_agy_relogin(cfg, ctx)
    if relogin.ok:
        return ActionResult(True, "설치가 끝났습니다. 이어서 로그인 창을 열었습니다.",
                            verify_check_key="agy_auth")
    return ActionResult(True, "설치가 끝났습니다. [로그인 창 열기]를 눌러 "
                              "Google 계정 로그인을 진행하세요.",
                        verify_check_key="agy_auth")

scripts/bootstrap_agy.ps1 전문

# scripts/bootstrap_agy.ps1
# agy 존재 확인 -> 미설치 시 공식 install.ps1 무인 실행 -> 버전 출력
# 종료 코드: 0 = 사용 가능, 1 = 실패
[CmdletBinding()]
param(
    [switch]$Force   # 이미 설치돼 있어도 재설치
)

$ErrorActionPreference = 'Stop'
$agyExe = Join-Path $env:LOCALAPPDATA 'agy\bin\agy.exe'

# 설치 중 자동 업데이터가 끼어들지 않게 한다 (agy SSOT §3.4)
$env:AGY_CLI_DISABLE_AUTO_UPDATE = 'true'

function Test-Agy {
    param([string]$Path)
    if (-not (Test-Path $Path)) { return $false }
    try {
        $v = & $Path --version 2>&1
        Write-Host "agy 확인됨: $Path ($v)"
        return $true
    } catch {
        return $false
    }
}

if ((Test-Agy -Path $agyExe) -and (-not $Force)) {
    exit 0
}

if ($Force -and (Test-Path $agyExe)) {
    # install.ps1 은 기존 설치를 감지하면 아무것도 하지 않고 0 으로 빠진다
    # (agy SSOT §3.3 3단계). 재설치하려면 바이너리를 먼저 지워야 한다.
    Write-Host '기존 agy 바이너리를 제거합니다(재설치 요청).'
    Remove-Item $agyExe -Force
}

Write-Host 'Antigravity CLI 를 설치합니다. 인터넷 연결이 필요합니다...'
try {
    [Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12
    $script = Invoke-RestMethod -Uri 'https://antigravity.google/cli/install.ps1' -TimeoutSec 60
    Invoke-Expression $script
} catch {
    Write-Error "설치 스크립트 실행에 실패했습니다: $($_.Exception.Message)"
    exit 1
}

if (Test-Agy -Path $agyExe) {
    Write-Host '설치가 완료됐습니다.'
    Write-Host '최초 1회 Google 계정 로그인이 필요합니다.'
    exit 0
}

Write-Error "설치는 끝났지만 $agyExe 를 찾을 수 없습니다."
exit 1

where.exe agy 를 쓰지 않는 이유: agy SSOT §3.2 실측에 따르면 winget 설치 이력이 있으면 %LOCALAPPDATA%\Microsoft\WinGet\Links\agy.EXE 가 함께 잡힌다. 배치 스크립트는 절대 경로를 고정해서 쓴다.

6. 알림 스크립트 전문

6.0 파일 구성과 아키텍처 트리 증분 (AMD-02)

파일	상태	역할
`src/dmf_crawler/notify/toast.py`	아키텍처 트리에 있음	tkinter 자동소멸 토스트
`src/dmf_crawler/notify/eventlog.py`	아키텍처 트리에 있음	`eventcreate.exe` 래퍼
`src/dmf_crawler/notify/messages.py`	아키텍처 트리에 있음	문구 템플릿(§3.4)
`src/dmf_crawler/notify/pump.py`	아키텍처 트리에 있음	대화형 에이전트 본체
`src/dmf_crawler/notify/webhook.py`	신규(AMD-02)	웹훅·dead-man switch. 선택 기능이므로 파일 하나로 격리
`scripts/notify.ps1`	신규(AMD-02)	Python 계층이 죽었을 때의 최후 알림기
`scripts/register_protocol.ps1`	신규(AMD-02)	`dmf://` 등록(§4.4). 기본 경로에서는 실행하지 않음

scripts/notify.ps1 을 신설하는 근거: 실패 시나리오 #32(Python/venv 손상)에서 tkinter 토스트는 원리적으로 뜰 수 없다. 아키텍처는 이 경우 "Event Log 가 유일한 흔적"이라고 적었는데, 이벤트 로그는 사용자가 보러 가지 않으면 아무 소용이 없다. PowerShell 은 Windows 에 내장돼 있고 우리 venv 와 독립적이므로, 이 하나의 시나리오를 위해 파일 하나를 추가할 가치가 있다.

6.1 `scripts/notify.ps1` 전문

<#
.SYNOPSIS
    DMF 크롤러 최후 알림기 (Python 독립).

.DESCRIPTION
    Python/venv 가 손상돼 notify/pump.py 가 돌지 못할 때를 위한 폴백 알림기다.
    DMF_Crawler_Agent 작업의 "두 번째 액션"으로 등록되어, pump 가 스탬프를
    남기지 못했을 때만 화면에 뜬다. 정상 상태에서는 아무것도 하지 않는다.

    표시 사다리:
        1) System.Windows.Forms.MessageBox  (기본)
        2) msg.exe * <메시지>               (.NET 로드 실패 시)
        3) Windows 이벤트 로그              (항상 병행)

.PARAMETER Mode
    Guard  : pump 스탬프 나이를 검사해 필요할 때만 알린다 (스케줄러가 쓰는 모드)
    Force  : 무조건 state/alerts.json 의 미해소 알림을 표시한다 (테스트용)
    Test   : 더미 알림 1건을 표시한다 (설치 검증용)

.PARAMETER ProjectRoot
    프로젝트 루트. 기본값은 이 스크립트의 부모 디렉터리.

.EXAMPLE
    powershell -NoProfile -ExecutionPolicy Bypass -File scripts\notify.ps1 -Mode Guard
#>
[CmdletBinding()]
param(
    [ValidateSet('Guard', 'Force', 'Test')]
    [string]$Mode = 'Guard',

    [string]$ProjectRoot = (Split-Path -Parent $PSScriptRoot),

    [int]$StaleMinutes = 45
)

$ErrorActionPreference = 'Continue'   # 알림기는 절대 죽지 않는다
$EventSource = 'DMF Crawler'

# ------------------------------------------------------------------ 경로
$StateDir   = Join-Path $ProjectRoot 'state'
$StampPath  = Join-Path $StateDir 'pump.stamp'
$AlertsPath = Join-Path $StateDir 'alerts.json'
$BootstrapPath = Join-Path $ProjectRoot 'bootstrap.cmd'

# ------------------------------------------------------- 이벤트 로그 기록
function Write-DmfEvent {
    param(
        [Parameter(Mandatory)][int]$Id,
        [ValidateSet('INFORMATION', 'WARNING', 'ERROR', 'SUCCESS')]
        [string]$Type = 'WARNING',
        [Parameter(Mandatory)][string]$Message
    )
    # eventcreate.exe 는 /ID 를 1~1000 으로만 받는다.
    if ($Id -lt 1 -or $Id -gt 1000) { $Id = 910 }
    # /D 는 명령줄 길이 제한이 있다. 안전하게 자른다.
    $desc = $Message -replace '\r?\n', ' | '
    if ($desc.Length -gt 900) { $desc = $desc.Substring(0, 900) + '...' }
    try {
        & eventcreate.exe /L APPLICATION /SO $EventSource /T $Type /ID $Id /D $desc 2>&1 |
            Out-Null
    } catch {
        # 이벤트 로그 기록 실패는 무시한다. 화면 알림이 본체다.
    }
}

# --------------------------------------------------------- 표시 사다리 1단
function Show-MessageBoxAlert {
    param(
        [Parameter(Mandatory)][string]$Title,
        [Parameter(Mandatory)][string]$Body,
        [ValidateSet('Error', 'Warning', 'Information')]
        [string]$Icon = 'Error'
    )
    try {
        Add-Type -AssemblyName System.Windows.Forms -ErrorAction Stop
        Add-Type -AssemblyName System.Drawing -ErrorAction Stop
    } catch {
        return $false
    }
    try {
        # MB_SYSTEMMODAL 에 해당하는 TopMost 를 주기 위해 더미 폼을 소유자로 쓴다.
        # 이것이 없으면 다른 창 뒤로 숨어 사용자가 영영 못 본다.
        $owner = New-Object System.Windows.Forms.Form
        $owner.TopMost      = $true
        $owner.ShowInTaskbar = $false
        $owner.StartPosition = 'CenterScreen'
        $owner.Size          = New-Object System.Drawing.Size(1, 1)
        $owner.Opacity       = 0
        $owner.Show()

        [void][System.Windows.Forms.MessageBox]::Show(
            $owner,
            $Body,
            $Title,
            [System.Windows.Forms.MessageBoxButtons]::OK,
            [System.Windows.Forms.MessageBoxIcon]::$Icon,
            [System.Windows.Forms.MessageBoxDefaultButton]::Button1
        )
        $owner.Close()
        $owner.Dispose()
        return $true
    } catch {
        return $false
    }
}

# --------------------------------------------------------- 표시 사다리 2단
function Show-MsgExeAlert {
    param([Parameter(Mandatory)][string]$Body)
    # msg.exe 는 Windows Home 에디션에 없는 경우가 있다. 존재부터 확인한다.
    $msg = Get-Command msg.exe -ErrorAction SilentlyContinue
    if (-not $msg) { return $false }
    try {
        # 한 줄로 눌러 보낸다. msg.exe 는 개행을 잘 다루지 못한다.
        $flat = ($Body -replace '\r?\n', '  ')
        if ($flat.Length -gt 250) { $flat = $flat.Substring(0, 250) + '...' }
        & $msg.Source '*' '/TIME:120' $flat 2>&1 | Out-Null
        return $true
    } catch {
        return $false
    }
}

# ------------------------------------------------------------ 표시 오케스트레이션
function Invoke-Alert {
    param(
        [Parameter(Mandatory)][string]$Title,
        [Parameter(Mandatory)][string]$Body,
        [int]$EventId = 910,
        [string]$EventType = 'ERROR'
    )
    Write-DmfEvent -Id $EventId -Type $EventType -Message "$Title | $Body"

    if (Show-MessageBoxAlert -Title $Title -Body $Body -Icon Error) {
        Write-Host "[notify.ps1] MessageBox 로 표시했습니다."
        return
    }
    Write-DmfEvent -Id 510 -Type 'WARNING' `
        -Message 'MessageBox 표시 실패. msg.exe 로 강등합니다.'

    if (Show-MsgExeAlert -Body "$Title`n$Body") {
        Write-Host "[notify.ps1] msg.exe 로 표시했습니다."
        return
    }
    Write-DmfEvent -Id 510 -Type 'ERROR' `
        -Message '모든 화면 알림 채널이 실패했습니다. 이벤트 로그만 남습니다.'
    Write-Host "[notify.ps1] 화면 표시에 모두 실패했습니다."
}

# --------------------------------------------------------------- 모드별 동작
function Get-StampAgeMinutes {
    if (-not (Test-Path $StampPath)) { return [int]::MaxValue }
    try {
        $mtime = (Get-Item $StampPath).LastWriteTime
        return [int]((Get-Date) - $mtime).TotalMinutes
    } catch {
        return [int]::MaxValue
    }
}

function Invoke-GuardMode {
    $age = Get-StampAgeMinutes
    if ($age -le $StaleMinutes) {
        # 정상. 아무것도 하지 않는다. 이것이 대부분의 실행 경로다.
        Write-Host "[notify.ps1] 정상 (스탬프 ${age}분 전). 아무것도 하지 않습니다."
        return 0
    }

    $lastText = if (Test-Path $StampPath) {
        (Get-Item $StampPath).LastWriteTime.ToString('yyyy-MM-dd HH:mm')
    } else {
        '기록 없음'
    }

    $title = 'DMF 크롤러 - 프로그램이 실행되지 않습니다'
    $body  = @"
[무엇]   DMF 크롤러의 알림 프로그램이 ${age}분째 응답하지 않습니다.
         (마지막 정상 동작: $lastText)

[왜]     Python 실행 환경(.venv)이 손상됐거나 삭제됐을 가능성이 큽니다.
         이 상태에서는 매일 06:00 자동 수집도 함께 멈춥니다.

[어떻게] 아래 폴더의 bootstrap.cmd 를 더블클릭해 다시 설치하세요(약 3분).
         기존 데이터와 설정은 그대로 유지됩니다.

폴더: $ProjectRoot
설치: $BootstrapPath

* 자세한 기록은 이벤트 뷰어 > Windows 로그 > 응용 프로그램에서
  원본 "DMF Crawler" 로 확인할 수 있습니다.
"@
    Invoke-Alert -Title $title -Body $body -EventId 910 -EventType 'ERROR'

    # 폴더를 함께 열어 준다. 사용자가 경로를 타이핑하지 않아도 되게.
    try { Start-Process explorer.exe -ArgumentList $ProjectRoot } catch { }
    return 1
}

function Invoke-ForceMode {
    if (-not (Test-Path $AlertsPath)) {
        Write-Host "[notify.ps1] $AlertsPath 가 없습니다."
        return 0
    }
    try {
        $data = Get-Content -Path $AlertsPath -Raw -Encoding UTF8 | ConvertFrom-Json
    } catch {
        Invoke-Alert -Title 'DMF 크롤러 - 알림 파일을 읽지 못했습니다' `
                     -Body "state\alerts.json 이 손상됐습니다.`n경로: $AlertsPath" `
                     -EventId 910
        return 1
    }

    $pending = @($data.pending | Where-Object { $_.severity -in @('ERROR', 'CRITICAL') })
    if ($pending.Count -eq 0) {
        Write-Host "[notify.ps1] 표시할 ERROR/CRITICAL 알림이 없습니다."
        return 0
    }

    foreach ($a in $pending) {
        $body = @"
[무엇]   $($a.what)

[왜]     $($a.why)

[어떻게] $($a.how)

발생: $($a.last_seen_at)   (누적 $($a.occurrences)회)
로그: $($a.log_dir)
"@
        Invoke-Alert -Title "DMF 크롤러 - $($a.title)" -Body $body `
                     -EventId 400 -EventType 'ERROR'
    }
    return 1
}

function Invoke-TestMode {
    $body = @"
[무엇]   이것은 알림 채널 점검용 시험 메시지입니다.

[왜]     scripts\notify.ps1 -Mode Test 로 직접 실행했습니다.

[어떻게] 이 창이 보인다면 폴백 알림 채널이 정상입니다. [확인]을 누르세요.

프로젝트: $ProjectRoot
표시 시각: $(Get-Date -Format 'yyyy-MM-dd HH:mm:ss')
"@
    Invoke-Alert -Title 'DMF 크롤러 - 알림 시험' -Body $body `
                 -EventId 500 -EventType 'INFORMATION'
    return 0
}

# ------------------------------------------------------------------- 진입점
switch ($Mode) {
    'Guard' { exit (Invoke-GuardMode) }
    'Force' { exit (Invoke-ForceMode) }
    'Test'  { exit (Invoke-TestMode) }
}

Agent 작업에 2번째 액션으로 등록 (scripts/install_tasks.ps1 발췌)

# DMF_Crawler_Agent — 액션 2개. 순서대로 실행된다.
$agentActions = @(
    # 1) 정상 경로: Python 알림 펌프. 콘솔 창이 뜨지 않는다.
    (New-ScheduledTaskAction -Execute $PythonwExe `
        -Argument '-m dmf_crawler notify-pump --once' `
        -WorkingDirectory $ProjectRoot),

    # 2) 폴백 경로: pump 가 스탬프를 못 남겼을 때만 동작한다.
    #    정상 상태에서는 즉시 종료되므로 부담이 없다.
    (New-ScheduledTaskAction -Execute 'powershell.exe' `
        -Argument ("-NoProfile -NonInteractive -WindowStyle Hidden " +
                   "-ExecutionPolicy Bypass -File `"$ProjectRoot\scripts\notify.ps1`" " +
                   "-Mode Guard -StaleMinutes 45") `
        -WorkingDirectory $ProjectRoot)
)

$agentTrigger = @(
    (New-ScheduledTaskTrigger -AtLogOn -User $TargetUser),
    (New-ScheduledTaskTrigger -Once -At (Get-Date).Date.AddMinutes(1) `
        -RepetitionInterval (New-TimeSpan -Minutes 15))
)

# LogonType Interactive 가 절대적으로 중요하다. 이것이 UI 를 띄울 수 있는 유일한 조건이다.
$agentPrincipal = New-ScheduledTaskPrincipal -UserId $TargetUser `
    -LogonType Interactive -RunLevel Limited

$agentSettings = New-ScheduledTaskSettingsSet `
    -AllowStartIfOnBatteries -DontStopIfGoingOnBatteries `
    -StartWhenAvailable -MultipleInstances IgnoreNew `
    -ExecutionTimeLimit (New-TimeSpan -Minutes 10) -Priority 7

Register-ScheduledTask -TaskName 'DMF_Crawler_Agent' `
    -Action $agentActions -Trigger $agentTrigger `
    -Principal $agentPrincipal -Settings $agentSettings -Force | Out-Null

작업 스케줄러의 다중 액션은 순차 실행이며, 앞 액션의 종료 코드를 보지 않는다. 즉 pump 가 죽어도 2번 액션은 반드시 실행된다 — 이것이 폴백이 성립하는 이유다. 반대로 pump 가 정상이면 notify.ps1 -Mode Guard 는 스탬프를 보고 즉시 0으로 빠지므로 중복 알림이 나지 않는다.

6.2 `src/dmf_crawler/notify/toast.py` 전문

# src/dmf_crawler/notify/toast.py
"""tkinter 로 그린 우하단 자동소멸 알림 창.

왜 tkinter 인가 (ADR-12):
    - 표준 라이브러리다. 의존성이 늘지 않는다.
    - 집중 지원(방해 금지) 모드의 영향을 받지 않는다. OS 알림 API 를 쓰지 않기 때문.
    - BurntToast/win11toast 는 외부 설치가 전제여서,
      "설치가 깨지면 알림도 안 뜬다"는 순환 실패를 만든다.

한계 (정직하게 기록):
    - 액션 센터에 남지 않는다. 놓치면 사라진다.
      -> 그래서 alerts 테이블과 state/alerts.json 이 원본이고 이 창은 사본일 뿐이다.
    - 배타적 전체 화면 앱 위에는 뜨지 못한다. -> is_presentation_mode() 로 회피한다.
"""
from __future__ import annotations

import ctypes
import tkinter as tk
from dataclasses import dataclass
from typing import Callable, Sequence

# --------------------------------------------------------------- 디자인 토큰
# 리포트와 같은 Okabe-Ito 계열. 색각이상 안전.
_BG        = "#1B1D23"
_FG        = "#F2F3F5"
_FG_DIM    = "#A8ADB7"
_BORDER    = "#3A3F4B"
_ACCENT = {
    "INFO":     "#0072B2",   # 파랑
    "WARN":     "#E69F00",   # 주황
    "ERROR":    "#D55E00",   # 주황빨강
    "CRITICAL": "#CC3311",   # 빨강
}
_BTN_BG      = "#2A2E38"
_BTN_BG_HOVER = "#3A3F4B"
_FONT_FAMILY = "맑은 고딕"

_MARGIN_RIGHT  = 24
_MARGIN_BOTTOM = 56    # 작업 표시줄을 피한다
_WIDTH         = 440
_GAP           = 12    # 토스트 여러 장을 쌓을 때의 간격

# 이미 떠 있는 토스트들의 높이 누적(스택 배치용)
_stack_offset = 0


@dataclass(frozen=True, slots=True)
class ToastButton:
    key: str
    label: str


@dataclass(frozen=True, slots=True)
class ToastResult:
    clicked: str | None      # 눌린 버튼의 key. 자동 소멸이면 None
    shown: bool              # 창이 실제로 화면에 그려졌는가


def enable_dpi_awareness() -> None:
    """고DPI 화면에서 창이 흐려지는 것을 막는다. 실패해도 무시한다."""
    try:
        # PROCESS_PER_MONITOR_DPI_AWARE = 2
        ctypes.windll.shcore.SetProcessDpiAwareness(2)
    except Exception:
        try:
            ctypes.windll.user32.SetProcessDPIAware()
        except Exception:
            pass


def is_presentation_mode() -> bool:
    """전체 화면 앱·프레젠테이션 모드인지 확인한다(B4 대응).

    SHQueryUserNotificationState 반환값:
        1 NOT_PRESENT        2 BUSY                 3 RUNNING_D3D_FULL_SCREEN
        4 PRESENTATION_MODE  5 ACCEPTS_NOTIFICATIONS 6 QUIET_TIME
        7 APP  (Windows 8+: 전체 화면 앱)
    """
    try:
        state = ctypes.c_int(0)
        hr = ctypes.windll.shell32.SHQueryUserNotificationState(ctypes.byref(state))
        if hr != 0:
            return False
        return state.value in (2, 3, 4, 7)
    except Exception:
        return False


def reset_stack() -> None:
    """pump 주기 시작 시 호출. 토스트 쌓임 위치를 초기화한다."""
    global _stack_offset
    _stack_offset = 0


def show(
    *,
    title: str,
    body: str,
    severity: str = "WARN",
    buttons: Sequence[ToastButton] = (),
    seconds: int = 12,
    on_click: Callable[[str], None] | None = None,
) -> ToastResult:
    """토스트를 띄우고 닫힐 때까지 블록한다.

    반환:
        ToastResult(clicked=눌린 버튼 키 또는 None, shown=실제로 그려졌는지)

    이 함수는 절대 예외를 밖으로 던지지 않는다. 실패하면 shown=False 로 알린다.
    """
    global _stack_offset
    try:
        return _show_impl(title, body, severity, tuple(buttons), seconds, on_click)
    except Exception:
        return ToastResult(clicked=None, shown=False)


def _show_impl(
    title: str,
    body: str,
    severity: str,
    buttons: tuple[ToastButton, ...],
    seconds: int,
    on_click: Callable[[str], None] | None,
) -> ToastResult:
    global _stack_offset

    enable_dpi_awareness()
    accent = _ACCENT.get(severity.upper(), _ACCENT["WARN"])
    clicked: dict[str, str | None] = {"key": None}

    root = tk.Tk()
    root.withdraw()

    win = tk.Toplevel(root)
    win.overrideredirect(True)          # 제목 표시줄 없음
    win.attributes("-topmost", True)
    win.configure(bg=_BORDER)           # 바깥 1px 테두리 역할

    # ------------------------------------------------------------ 레이아웃
    outer = tk.Frame(win, bg=_BG)
    outer.pack(padx=1, pady=1, fill="both", expand=True)

    # 왼쪽 등급 색 띠
    tk.Frame(outer, bg=accent, width=5).pack(side="left", fill="y")

    inner = tk.Frame(outer, bg=_BG)
    inner.pack(side="left", fill="both", expand=True, padx=16, pady=14)

    tk.Label(
        inner, text=title, bg=_BG, fg=_FG, justify="left", anchor="w",
        font=(_FONT_FAMILY, 11, "bold"), wraplength=_WIDTH - 60,
    ).pack(fill="x")

    tk.Label(
        inner, text=body, bg=_BG, fg=_FG_DIM, justify="left", anchor="w",
        font=(_FONT_FAMILY, 9), wraplength=_WIDTH - 60,
    ).pack(fill="x", pady=(8, 0))

    # ------------------------------------------------------------- 버튼들
    def _close(key: str | None) -> None:
        clicked["key"] = key
        try:
            win.destroy()
            root.quit()
        except tk.TclError:
            pass

    if buttons:
        bar = tk.Frame(inner, bg=_BG)
        bar.pack(fill="x", pady=(14, 0))
        for btn in buttons:
            b = tk.Button(
                bar, text=btn.label, bg=_BTN_BG, fg=_FG,
                activebackground=_BTN_BG_HOVER, activeforeground=_FG,
                relief="flat", bd=0, padx=12, pady=5, cursor="hand2",
                font=(_FONT_FAMILY, 9),
                command=lambda k=btn.key: _on_button(k, on_click, _close),
            )
            b.pack(side="left", padx=(0, 8))
            b.bind("<Enter>", lambda e, w=b: w.configure(bg=_BTN_BG_HOVER))
            b.bind("<Leave>", lambda e, w=b: w.configure(bg=_BTN_BG))

    # ------------------------------------------------- 위치 계산(우하단 스택)
    win.update_idletasks()
    height = win.winfo_reqheight()
    screen_w = win.winfo_screenwidth()
    screen_h = win.winfo_screenheight()
    x = screen_w - _WIDTH - _MARGIN_RIGHT
    y = screen_h - _MARGIN_BOTTOM - height - _stack_offset
    if y < 40:                      # 화면 위로 넘치면 스택을 접는다
        _stack_offset = 0
        y = screen_h - _MARGIN_BOTTOM - height
    win.geometry(f"{_WIDTH}x{height}+{x}+{y}")
    _stack_offset += height + _GAP

    # --------------------------------------------------------- 동작 바인딩
    # 본문 클릭 = 첫 번째 버튼과 같은 동작(가장 흔한 조작을 쉽게)
    default_key = buttons[0].key if buttons else None
    for widget in (inner, outer):
        widget.bind("<Button-1>",
                    lambda e: _on_button(default_key, on_click, _close)
                    if default_key else _close(None))
    # 오른쪽 클릭 = 그냥 닫기
    win.bind("<Button-3>", lambda e: _close(None))
    win.bind("<Escape>", lambda e: _close(None))

    # 자동 소멸 타이머
    timer_id = win.after(max(1, seconds) * 1000, lambda: _close(None))

    # 마우스를 올리면 타이머를 멈춘다(읽는 중에 사라지면 안 된다)
    def _pause(_e: object) -> None:
        try:
            win.after_cancel(timer_id)
        except (tk.TclError, ValueError):
            pass

    def _resume(_e: object) -> None:
        nonlocal timer_id
        try:
            timer_id = win.after(3000, lambda: _close(None))
        except tk.TclError:
            pass

    win.bind("<Enter>", _pause)
    win.bind("<Leave>", _resume)

    # 페이드 인
    try:
        win.attributes("-alpha", 0.0)
        for step in range(0, 11):
            win.attributes("-alpha", step / 10.0 * 0.97)
            win.update()
            win.after(12)
    except tk.TclError:
        pass

    root.mainloop()
    try:
        root.destroy()
    except tk.TclError:
        pass

    return ToastResult(clicked=clicked["key"], shown=True)


def _on_button(
    key: str | None,
    on_click: Callable[[str], None] | None,
    close: Callable[[str | None], None],
) -> None:
    if key and on_click:
        try:
            on_click(key)
        except Exception:
            pass          # 액션 실패가 창을 죽이면 안 된다
    close(key)


def show_merged(
    *,
    count: int,
    lines: Sequence[str],
    severity: str,
    seconds: int,
    on_click: Callable[[str], None] | None = None,
) -> ToastResult:
    """여러 알림을 한 장으로 묶어 표시한다(L2 병합)."""
    body = "\n".join(f"  · {line}" for line in lines)
    return show(
        title=f"DMF 크롤러 — 확인이 필요한 항목 {count}건",
        body=body,
        severity=severity,
        buttons=(
            ToastButton("open_doctor", "자세히 보기"),
            ToastButton("open_report_file", "리포트 열기"),
            ToastButton("dismiss", "닫기"),
        ),
        seconds=seconds,
        on_click=on_click,
    )

6.3 `src/dmf_crawler/notify/eventlog.py` 전문

# src/dmf_crawler/notify/eventlog.py
"""Windows 이벤트 로그 병행 기록.

pywin32 를 쓰지 않는다(ADR 의존성 최소주의). Windows 내장 eventcreate.exe 를 부른다.

알려진 제약 (반드시 지킬 것):
    1. /ID 는 1~1000 범위만 허용된다. 벗어나면 명령 자체가 실패한다.
       -> EVENT_IDS 상수가 이 범위를 강제한다.
    2. /D 는 명령줄 인자이므로 길이 제한이 있다. 900자로 자른다.
    3. /SO 로 지정한 원본은 Application 로그에 자동 등록된다.
       이미 시스템에 등록된 원본 이름과 충돌하면 실패하므로 고유한 이름을 쓴다.
    4. PowerShell 7 에는 Write-EventLog/New-EventLog 가 없다.
       eventcreate.exe 는 exe 이므로 셸 종류와 무관하게 동작한다.
"""
from __future__ import annotations

import subprocess
from typing import Literal

from dmf_crawler.alerts import Severity

CREATE_NO_WINDOW = 0x08000000

EventType = Literal["INFORMATION", "WARNING", "ERROR", "SUCCESS"]

# 등급 -> eventcreate /T 매핑
_TYPE_BY_SEVERITY: dict[str, EventType] = {
    "INFO": "INFORMATION",
    "WARN": "WARNING",
    "ERROR": "ERROR",
    "CRITICAL": "ERROR",   # eventcreate 에 CRITICAL 타입은 없다
}

# 등급 기본 ID (템플릿이 eventlog_id 를 지정하면 그것이 우선한다)
_ID_BY_SEVERITY: dict[str, int] = {
    "INFO": 100, "WARN": 200, "ERROR": 300, "CRITICAL": 400,
}

# 이 문서 §2.8 의 ID 배정표
EVENT_IDS: dict[str, int] = {
    "RUN_START": 100, "RUN_SUCCESS": 110, "RUN_PARTIAL": 120, "RUN_SKIPPED": 130,
    "ALERT_WARN": 200, "ALERT_ERROR": 300, "ALERT_CRITICAL": 400,
    "WATCHDOG_STALE": 410, "CONSECUTIVE_FAILURES": 420, "RUN_PAUSED": 430,
    "NOTIFY_SHOWN": 500, "NOTIFY_DEGRADED": 510, "NOTIFY_SUPPRESSED": 520,
    "PUMP_CRASHED": 900, "FALLBACK_FIRED": 910,
}

_MAX_DESC = 900


def write(
    *,
    source: str,
    message: str,
    severity: Severity | str = Severity.INFO,
    event_id: int | None = None,
    timeout_s: float = 10.0,
) -> bool:
    """이벤트 로그에 1건 기록한다. 실패해도 예외를 던지지 않고 False 를 반환한다."""
    sev = str(severity)
    etype: EventType = _TYPE_BY_SEVERITY.get(sev, "WARNING")
    eid = event_id if event_id else _ID_BY_SEVERITY.get(sev, 200)
    if not (1 <= eid <= 1000):     # eventcreate 의 하드 제약
        eid = _ID_BY_SEVERITY.get(sev, 200)

    desc = " | ".join(line.strip() for line in message.splitlines() if line.strip())
    if len(desc) > _MAX_DESC:
        desc = desc[: _MAX_DESC - 3] + "..."
    if not desc:
        desc = "(내용 없음)"

    try:
        proc = subprocess.run(
            ["eventcreate.exe",
             "/L", "APPLICATION",
             "/SO", source,
             "/T", etype,
             "/ID", str(eid),
             "/D", desc],
            capture_output=True, text=True, encoding="cp949", errors="replace",
            timeout=timeout_s, creationflags=CREATE_NO_WINDOW,
        )
    except (OSError, subprocess.TimeoutExpired):
        return False
    return proc.returncode == 0


def write_alert(cfg, alert_code: str, severity: Severity, title: str,
                body: str, event_id: int | None = None) -> bool:
    """알림 1건을 이벤트 로그에 남긴다."""
    return write(
        source=cfg.notify.eventlog_source,
        message=f"[{alert_code}] {title} :: {body}",
        severity=severity,
        event_id=event_id,
    )

수동 검증

# 기록
eventcreate.exe /L APPLICATION /SO "DMF Crawler" /T ERROR /ID 400 /D "테스트 알림"

# 확인
Get-WinEvent -FilterHashtable @{LogName='Application'; ProviderName='DMF Crawler'} -MaxEvents 5 |
    Format-List TimeCreated, Id, LevelDisplayName, Message

6.4 `src/dmf_crawler/alerts.py` 전문

# src/dmf_crawler/alerts.py
"""알림 의도의 기록·중복 억제·표시 추적.

원칙 (ADR-11):
    이 모듈은 절대로 화면에 무엇을 띄우지 않는다. 기록만 한다.
    표시는 notify/pump.py 가 로그온 세션에서 한다.

계약 (요구 R7.3):
    raise_alert 는 what/why/how/next_actions 중 하나라도 비면 ValueError 를 던진다.
"""
from __future__ import annotations

import json
import os
import socket
import sqlite3
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import StrEnum
from pathlib import Path
from typing import Any, Mapping, Sequence


class Severity(StrEnum):
    INFO     = "INFO"
    WARN     = "WARN"
    ERROR    = "ERROR"
    CRITICAL = "CRITICAL"

    @property
    def rank(self) -> int:
        return {"INFO": 0, "WARN": 1, "ERROR": 2, "CRITICAL": 3}[self.value]

    def at_least(self, other: "Severity") -> bool:
        return self.rank >= other.rank


@dataclass(frozen=True, slots=True)
class AlertRow:
    alert_id: int
    dedup_key: str
    code: str
    severity: Severity
    first_seen_at: str
    last_seen_at: str
    occurrences: int
    show_attempts: int
    title: str
    what: str
    why: str
    how: str
    next_actions: tuple[str, ...]
    context: dict[str, Any]
    log_dir: str | None
    run_id: str | None

    def as_context(self) -> dict[str, Any]:
        ctx = dict(self.context)
        ctx.update({
            "alert_id": self.alert_id,
            "code": self.code,
            "occurrences": self.occurrences,
            "log_dir": self.log_dir,
            "run_id": self.run_id,
        })
        return ctx

    @property
    def body(self) -> str:
        lines = [f"[무엇]   {self.what}",
                 f"[왜]     {self.why}",
                 f"[어떻게] {self.how}"]
        if self.log_dir:
            lines += ["", f"로그: {self.log_dir}"]
        return "\n".join(lines)


def _now_iso() -> str:
    return datetime.now().astimezone().isoformat(timespec="seconds")


def make_dedup_key(code: str, run_date: str | None,
                   scope: str | None = None) -> str:
    parts = [code]
    if run_date:
        parts.append(run_date)
    if scope:
        parts.append(scope)
    return "|".join(parts)


def raise_alert(
    conn: sqlite3.Connection,
    *,
    run_id: str | None,
    severity: Severity,
    code: str,
    what: str,
    why: str,
    how: str,
    next_actions: Sequence[str],
    context: Mapping[str, Any] | None = None,
    log_dir: Path | None = None,
    cooldown_minutes: int,
    dedup_scope: str | None = None,
    date_scoped: bool = True,
    source: str = "pipeline",
) -> bool:
    """알림 의도를 기록한다.

    반환:
        True  -> 새 알림이거나 쿨다운이 지나 표시 대상이 됐다
        False -> 쿨다운 안이라 카운터만 올렸다
    """
    # ---- 4요소 계약 강제 (요구 R7.3) --------------------------------
    for name, value in (("what", what), ("why", why), ("how", how)):
        if not value or not value.strip():
            raise ValueError(f"알림 {code}: '{name}' 이 비었습니다. "
                             f"4요소(무엇/왜/어떻게/다음 행동)는 필수입니다.")
    actions = tuple(a for a in next_actions if a)
    if not actions:
        raise ValueError(f"알림 {code}: next_actions 가 비었습니다. "
                         f"막다른 골목 알림은 금지입니다(R7.7).")
    if severity.at_least(Severity.WARN) and set(actions) <= {"snooze", "dismiss"}:
        raise ValueError(f"알림 {code}: 실행 가능한 액션이 없습니다(R7.7).")

    now = _now_iso()
    run_date = now[:10] if date_scoped else None
    key = make_dedup_key(code, run_date, dedup_scope)
    ctx_json = json.dumps(dict(context or {}), ensure_ascii=False)
    log_dir_str = str(log_dir) if log_dir else None

    cur = conn.cursor()
    # 1) 발생 사실은 무조건 append
    cur.execute(
        """INSERT INTO alert_events
               (occurred_at, run_id, code, severity, dedup_key,
                what, why, how, next_actions, context_json, log_dir, source)
           VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""",
        (now, run_id, code, str(severity), key,
         what, why, how, json.dumps(list(actions), ensure_ascii=False),
         ctx_json, log_dir_str, source),
    )
    event_id = cur.lastrowid

    # 2) 상태 행 조회
    cur.execute(
        """SELECT alert_id, last_seen_at, occurrences, shown_at, resolved_at
             FROM alerts WHERE dedup_key = ?""", (key,))
    row = cur.fetchone()

    if row is None or row["resolved_at"] is not None:
        # 새 알림, 또는 이미 해소된 뒤 재발생 -> 새 상태로 시작
        cur.execute(
            """INSERT INTO alerts
                   (dedup_key, code, severity, first_seen_at, last_seen_at,
                    occurrences, last_event_id, show_attempts)
               VALUES (?,?,?,?,?,1,?,0)
               ON CONFLICT(dedup_key) DO UPDATE SET
                    severity      = excluded.severity,
                    last_seen_at  = excluded.last_seen_at,
                    occurrences   = alerts.occurrences + 1,
                    last_event_id = excluded.last_event_id,
                    shown_at      = NULL,
                    shown_channel = NULL,
                    show_attempts = 0,
                    snoozed_until = NULL,
                    resolved_at   = NULL,
                    resolved_note = NULL""",
            (key, code, str(severity), now, now, event_id),
        )
        conn.commit()
        return True

    # 3) 기존 미해소 알림 -> 쿨다운 판정
    last_seen = datetime.fromisoformat(row["last_seen_at"])
    elapsed = datetime.now().astimezone() - last_seen
    within_cooldown = elapsed < timedelta(minutes=cooldown_minutes)

    if within_cooldown:
        cur.execute(
            """UPDATE alerts
                  SET last_seen_at = ?, occurrences = occurrences + 1,
                      last_event_id = ?, severity = ?
                WHERE alert_id = ?""",
            (now, event_id, str(severity), row["alert_id"]),
        )
        conn.commit()
        return False

    cur.execute(
        """UPDATE alerts
              SET last_seen_at = ?, occurrences = occurrences + 1,
                  last_event_id = ?, severity = ?,
                  shown_at = NULL, shown_channel = NULL, show_attempts = 0
            WHERE alert_id = ?""",
        (now, event_id, str(severity), row["alert_id"]),
    )
    conn.commit()
    return True


def pending(conn: sqlite3.Connection, *,
            modal_repeat_minutes: int = 60) -> list[AlertRow]:
    """표시 대기 알림 목록. 심각도 내림차순."""
    now = _now_iso()
    cur = conn.execute(
        """SELECT a.alert_id, a.dedup_key, a.code, a.severity,
                  a.first_seen_at, a.last_seen_at, a.occurrences, a.show_attempts,
                  e.what, e.why, e.how, e.next_actions, e.context_json,
                  e.log_dir, e.run_id
             FROM alerts a
             JOIN alert_events e ON e.event_id = a.last_event_id
            WHERE a.resolved_at IS NULL
              AND (a.snoozed_until IS NULL OR a.snoozed_until < :now)
              AND (a.shown_at IS NULL
                   OR (a.severity IN ('ERROR','CRITICAL')
                       AND julianday(:now) - julianday(a.shown_at)
                           > :repeat / 1440.0))
            ORDER BY CASE a.severity
                        WHEN 'CRITICAL' THEN 0 WHEN 'ERROR' THEN 1
                        WHEN 'WARN' THEN 2 ELSE 3 END,
                     a.last_seen_at DESC""",
        {"now": now, "repeat": modal_repeat_minutes},
    )
    from dmf_crawler.notify import messages   # 순환 방지를 위해 지연 임포트

    rows: list[AlertRow] = []
    for r in cur.fetchall():
        ctx = json.loads(r["context_json"] or "{}")
        tpl = messages.TEMPLATES.get(r["code"])
        title = tpl.title.format_map(messages._SafeDict(ctx)) if tpl else r["code"]
        rows.append(AlertRow(
            alert_id=r["alert_id"], dedup_key=r["dedup_key"], code=r["code"],
            severity=Severity(r["severity"]),
            first_seen_at=r["first_seen_at"], last_seen_at=r["last_seen_at"],
            occurrences=r["occurrences"], show_attempts=r["show_attempts"],
            title=title, what=r["what"], why=r["why"], how=r["how"],
            next_actions=tuple(json.loads(r["next_actions"])),
            context=ctx, log_dir=r["log_dir"], run_id=r["run_id"],
        ))
    return rows


def mark_shown(conn: sqlite3.Connection, alert_id: int, *, channel: str) -> None:
    conn.execute(
        """UPDATE alerts
              SET shown_at = ?, shown_channel = ?, show_attempts = show_attempts + 1
            WHERE alert_id = ?""",
        (_now_iso(), channel, alert_id))
    conn.commit()


def mark_show_failed(conn: sqlite3.Connection, alert_id: int) -> int:
    """표시 실패 카운터를 올리고 현재 값을 반환한다(R-P3 승격 판정용)."""
    conn.execute(
        "UPDATE alerts SET show_attempts = show_attempts + 1 WHERE alert_id = ?",
        (alert_id,))
    conn.commit()
    cur = conn.execute("SELECT show_attempts FROM alerts WHERE alert_id = ?",
                       (alert_id,))
    row = cur.fetchone()
    return int(row["show_attempts"]) if row else 0


def snooze(conn: sqlite3.Connection, alert_id: int, *, minutes: int) -> None:
    until = (datetime.now().astimezone() +
             timedelta(minutes=minutes)).isoformat(timespec="seconds")
    conn.execute(
        """UPDATE alerts
              SET snoozed_until = ?, shown_at = ?, shown_channel = 'modal',
                  show_attempts = show_attempts + 1
            WHERE alert_id = ?""",
        (until, _now_iso(), alert_id))
    conn.commit()


def resolve(conn: sqlite3.Connection, code: str, note: str) -> int:
    """해당 코드의 미해소 알림을 전부 해소한다. 해소한 건수를 반환한다."""
    cur = conn.execute(
        """UPDATE alerts SET resolved_at = ?, resolved_note = ?
            WHERE code = ? AND resolved_at IS NULL""",
        (_now_iso(), note, code))
    conn.commit()
    return cur.rowcount


def resolve_date_scoped(conn: sqlite3.Connection, note: str) -> int:
    """다음 실행이 성공했을 때 일자성 WARN 을 일괄 해소한다(R-D1)."""
    cur = conn.execute(
        """UPDATE alerts SET resolved_at = ?, resolved_note = ?
            WHERE resolved_at IS NULL
              AND severity IN ('INFO','WARN')
              AND instr(dedup_key, '|') > 0""",
        (_now_iso(), note))
    conn.commit()
    return cur.rowcount


def shown_today(conn: sqlite3.Connection) -> int:
    cur = conn.execute("SELECT shown_count FROM v_alert_shown_today")
    row = cur.fetchone()
    return int(row["shown_count"]) if row else 0


def mirror_to_file(conn: sqlite3.Connection, path: Path) -> None:
    """state/alerts.json 을 원자적으로 갱신한다.

    PowerShell 폴백(notify.ps1)과 GUI 가 SQLite 잠금 없이 읽는 유일한 경로다.
    실패해도 예외를 던지지 않는다.
    """
    try:
        rows = pending(conn)
        counts = {s.value: 0 for s in Severity}
        payload_rows = []
        for row in rows:
            counts[row.severity.value] += 1
            payload_rows.append({
                "alert_id": row.alert_id, "dedup_key": row.dedup_key,
                "code": row.code, "severity": row.severity.value,
                "first_seen_at": row.first_seen_at, "last_seen_at": row.last_seen_at,
                "occurrences": row.occurrences, "title": row.title,
                "what": row.what, "why": row.why, "how": row.how,
                "next_actions": list(row.next_actions),
                "log_dir": row.log_dir, "run_id": row.run_id,
            })
        payload = {
            "schema": 1,
            "updated_at": _now_iso(),
            "host": socket.gethostname(),
            "pending": payload_rows,
            "counts": counts,
        }
        path.parent.mkdir(parents=True, exist_ok=True)
        tmp = path.with_suffix(".json.tmp")
        tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2),
                       encoding="utf-8")
        os.replace(tmp, path)
    except Exception:
        pass

6.5 `src/dmf_crawler/watchdog.py` 전문

# src/dmf_crawler/watchdog.py
"""heartbeat 신선도 판정과 연속 실패 감시.

아키텍처 부록 결정: 워치독을 별도 작업으로 분리하지 않는다.
로그온 세션의 Agent(15분 주기) 안에서 판정하면 작업이 하나 줄고,
"알림이 뜨는 세션에서 판정한다"는 성질이 공짜로 따라온다.
"""
from __future__ import annotations

import json
import sqlite3
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path

from dmf_crawler import paths
from dmf_crawler.alerts import Severity, raise_alert
from dmf_crawler.config import Config


@dataclass(frozen=True, slots=True)
class Heartbeat:
    exists: bool
    last_success_at: datetime | None
    run_id: str | None
    status: str | None
    report_path: str | None
    log_dir: str | None

    @property
    def age_minutes(self) -> float:
        if self.last_success_at is None:
            return float("inf")
        delta = datetime.now().astimezone() - self.last_success_at
        return delta.total_seconds() / 60.0


def read_heartbeat(path: Path | None = None) -> Heartbeat:
    target = path or paths.HEARTBEAT_PATH
    try:
        data = json.loads(target.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return Heartbeat(False, None, None, None, None, None)
    try:
        ts = datetime.fromisoformat(data["last_success_at"])
    except (KeyError, ValueError):
        ts = None
    return Heartbeat(
        exists=True, last_success_at=ts,
        run_id=data.get("run_id"), status=data.get("status"),
        report_path=data.get("report_path"), log_dir=data.get("log_dir"),
    )


def write_heartbeat(cfg: Config, *, run_id: str, status: str,
                    report_path: str | None, log_dir: str) -> None:
    """성공/부분성공일 때만 호출한다. FAILED 에서는 절대 갱신하지 않는다.

    (심사에서 후보 C 가 '항상 갱신'해 dead-man switch 를 설계상 무력화한 것이
     최대 감점이었다. 여기서 같은 실수를 하면 워치독 전체가 무의미해진다.)
    """
    payload = {
        "last_success_at": datetime.now().astimezone().isoformat(timespec="seconds"),
        "run_id": run_id,
        "status": status,
        "report_path": report_path,
        "log_dir": log_dir,
    }
    paths.STATE_DIR.mkdir(parents=True, exist_ok=True)
    tmp = paths.HEARTBEAT_PATH.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    import os
    os.replace(tmp, paths.HEARTBEAT_PATH)


def touch_pump_stamp() -> None:
    """notify.ps1 폴백이 읽는 생존 신호. pump 가 매 주기 갱신한다."""
    paths.STATE_DIR.mkdir(parents=True, exist_ok=True)
    paths.PUMP_STAMP.write_text(
        datetime.now().astimezone().isoformat(timespec="seconds"), encoding="utf-8")


def consecutive_failed_days(conn: sqlite3.Connection) -> int:
    """오늘부터 거슬러 올라가며 연속 실패한 날 수를 센다."""
    cur = conn.execute(
        """SELECT substr(started_at, 1, 10) AS d,
                  MAX(CASE WHEN status IN ('SUCCESS','PARTIAL') THEN 1 ELSE 0 END) AS ok
             FROM runs
            WHERE started_at >= date('now', '-14 days')
            GROUP BY d
            ORDER BY d DESC""")
    days = 0
    for row in cur.fetchall():
        if row["ok"] == 1:
            break
        days += 1
    return days


def top_failure_causes(conn: sqlite3.Connection, limit: int = 3
                       ) -> list[tuple[str, int]]:
    """최근 14일 실패 알림 코드 상위 N개. CONSECUTIVE_FAILURES 문구에 쓴다."""
    cur = conn.execute(
        """SELECT code, COUNT(*) AS n
             FROM alert_events
            WHERE occurred_at >= datetime('now', '-14 days')
              AND severity IN ('ERROR','CRITICAL')
              AND code NOT IN ('CONSECUTIVE_FAILURES','RUN_PAUSED','WATCHDOG_STALE')
            GROUP BY code
            ORDER BY n DESC
            LIMIT ?""", (limit,))
    return [(r["code"], r["n"]) for r in cur.fetchall()]


def evaluate(conn: sqlite3.Connection, cfg: Config) -> list[str]:
    """워치독 판정 1회. 발생시킨 알림 코드 목록을 반환한다."""
    fired: list[str] = []
    hb = read_heartbeat()
    now = datetime.now().astimezone()

    # ---- 1) heartbeat 신선도 -------------------------------------------
    # 06:00 실행 + 여유 2시간. 08:00 이전에는 판정하지 않는다.
    scheduled_hour, scheduled_min = (int(x) for x in cfg.schedule.daily_time.split(":"))
    today_due = now.replace(hour=scheduled_hour, minute=scheduled_min,
                            second=0, microsecond=0)
    grace = timedelta(minutes=cfg.notify.watchdog_stale_minutes)
    if now >= today_due + grace and hb.age_minutes > cfg.notify.watchdog_stale_minutes:
        last_txt = (hb.last_success_at.strftime("%Y-%m-%d %H:%M")
                    if hb.last_success_at else "기록 없음")
        stale_h = (int(hb.age_minutes // 60)
                   if hb.last_success_at else "알 수 없음")
        if raise_alert(
            conn, run_id=None, severity=Severity.CRITICAL, code="WATCHDOG_STALE",
            what=f"지금 {now:%H:%M} 기준으로 오늘 {cfg.schedule.daily_time} 배치가 "
                 f"실행된 흔적이 없습니다. 마지막으로 성공한 실행은 "
                 f"{last_txt} ({stale_h}시간 전)입니다.",
            why="(1) PC가 실행 시각에 꺼져 있었고 아직 따라잡기가 실행되지 않음 "
                "(2) 작업 스케줄러 항목이 꺼졌거나 삭제됨 "
                f"(3) 실행이 제한 시간({cfg.schedule.execution_time_limit_minutes}분)을 "
                "넘겨 강제 종료됨",
            how="[지금 실행]을 누르면 즉시 오늘 자료를 수집합니다(약 2분). "
                "반복된다면 [작업 상태 확인]으로 스케줄러 등록 상태를 점검하세요.",
            next_actions=("run_now", "open_task_scheduler", "open_log_dir", "snooze"),
            context={"now_time": f"{now:%H:%M}", "last_success_at": last_txt,
                     "stale_hours": stale_h,
                     "exec_limit_min": cfg.schedule.execution_time_limit_minutes},
            log_dir=Path(hb.log_dir) if hb.log_dir else None,
            cooldown_minutes=cfg.notify.modal_repeat_minutes,
            date_scoped=False, source="watchdog",
        ):
            fired.append("WATCHDOG_STALE")

    # ---- 2) 연속 실패 --------------------------------------------------
    failed_days = consecutive_failed_days(conn)
    if failed_days >= cfg.notify.escalation_days:
        causes = top_failure_causes(conn, 3)
        while len(causes) < 3:
            causes.append(("(추가 원인 없음)", 0))
        ctx = {
            "failed_days": failed_days,
            "first_failed_date": (now - timedelta(days=failed_days - 1)).strftime("%Y-%m-%d"),
            "last_failed_date": now.strftime("%Y-%m-%d"),
            "top_cause_1": causes[0][0], "top_cause_1_count": causes[0][1],
            "top_cause_2": causes[1][0], "top_cause_2_count": causes[1][1],
            "top_cause_3": causes[2][0], "top_cause_3_count": causes[2][1],
            "pause_day": cfg.notify.escalation_stop_after_days,
        }
        from dmf_crawler.notify import messages
        rendered = messages.render("CONSECUTIVE_FAILURES", ctx)
        if raise_alert(
            conn, run_id=None, severity=Severity.CRITICAL,
            code="CONSECUTIVE_FAILURES",
            what=rendered.what, why=rendered.why, how=rendered.how,
            next_actions=rendered.next_actions, context=ctx,
            log_dir=Path(hb.log_dir) if hb.log_dir else None,
            cooldown_minutes=cfg.notify.modal_repeat_minutes,
            date_scoped=False, source="watchdog",
        ):
            fired.append("CONSECUTIVE_FAILURES")

    # ---- 3) 자동 실행 일시중지 (7일차) ----------------------------------
    if failed_days >= cfg.notify.escalation_stop_after_days:
        if not paths.PAUSE_FLAG.exists():
            paths.STATE_DIR.mkdir(parents=True, exist_ok=True)
            paths.PAUSE_FLAG.write_text(
                json.dumps({"paused_at": now.isoformat(timespec="seconds"),
                            "reason": f"{failed_days}일 연속 실패",
                            "failed_days": failed_days},
                           ensure_ascii=False), encoding="utf-8")
        ctx = {"failed_days": failed_days,
               "pause_at": now.strftime("%Y-%m-%d %H:%M"),
               "pause_flag_path": str(paths.PAUSE_FLAG)}
        from dmf_crawler.notify import messages
        rendered = messages.render("RUN_PAUSED", ctx)
        if raise_alert(
            conn, run_id=None, severity=Severity.CRITICAL, code="RUN_PAUSED",
            what=rendered.what, why=rendered.why, how=rendered.how,
            next_actions=rendered.next_actions, context=ctx,
            log_dir=Path(hb.log_dir) if hb.log_dir else None,
            cooldown_minutes=cfg.notify.modal_repeat_minutes,
            date_scoped=False, source="watchdog",
        ):
            fired.append("RUN_PAUSED")

    return fired

6.6 `src/dmf_crawler/notify/pump.py` 전문

# src/dmf_crawler/notify/pump.py
"""대화형 알림 에이전트 — UI 를 띄울 권한을 가진 유일한 프로세스.

실행 주체:
    Task Scheduler 작업 DMF_Crawler_Agent
    (LogonType Interactive, 15분 반복 + AtLogOn)

원칙:
    - 어떤 예외도 사용자에게 보이지 않게 삼킨다. 다만 이벤트 로그에는 반드시 남긴다.
      "알리미가 죽어서 조용해지는 것"이 이 시스템의 최악의 실패다.
    - 매 주기 pump.stamp 를 갱신한다. 이것이 없으면 notify.ps1 폴백이 발동한다.
"""
from __future__ import annotations

import sys
import traceback
from datetime import datetime

from dmf_crawler import checks, paths, watchdog
from dmf_crawler.alerts import AlertRow, Severity
from dmf_crawler.alerts import (mark_show_failed, mark_shown, mirror_to_file,
                                pending, snooze)
from dmf_crawler.config import Config, load_config
from dmf_crawler.gui import steps
from dmf_crawler.notify import eventlog, toast, webhook
from dmf_crawler.runlock import file_lock
from dmf_crawler.storage import db

_PUMP_LOCK = "pump.lock"


def pump_once(cfg: Config) -> int:
    """1회 주기. 종료 코드를 반환한다(0=정상, 1=표시 실패 있음, 2=치명)."""
    # 자체 락. 이전 주기의 모달이 아직 떠 있으면 새로 띄우지 않는다.
    try:
        with file_lock(paths.STATE_DIR / _PUMP_LOCK, timeout_s=0):
            return _pump_body(cfg)
    except TimeoutError:
        # 이미 다른 pump 가 돌고 있다. 정상적인 상황이다.
        watchdog.touch_pump_stamp()
        return 0


def _pump_body(cfg: Config) -> int:
    watchdog.touch_pump_stamp()
    toast.reset_stack()

    conn = db.connect(cfg, readonly=False)
    try:
        # ---- 1) 워치독 판정 -> 필요하면 alerts 에 기록 -------------------
        try:
            watchdog.evaluate(conn, cfg)
        except Exception:
            eventlog.write(source=cfg.notify.eventlog_source,
                           severity=Severity.ERROR, event_id=900,
                           message="워치독 판정 중 오류: " + traceback.format_exc(limit=3))

        # ---- 2) 자동 해소 판정 (R-D2) ------------------------------------
        _auto_resolve(conn, cfg)

        # ---- 3) 표시 대상 조회 -------------------------------------------
        rows = pending(conn, modal_repeat_minutes=cfg.notify.modal_repeat_minutes)
        mirror_to_file(conn, paths.ALERTS_MIRROR)
        if not rows:
            return 0

        # ---- 4) 일일 상한(L3) 판정 — CRITICAL 은 면제 ---------------------
        from dmf_crawler.alerts import shown_today
        used = shown_today(conn)
        criticals = [r for r in rows if r.severity is Severity.CRITICAL]
        others = [r for r in rows if r.severity is not Severity.CRITICAL]
        if used >= cfg.notify.daily_alert_cap and others:
            eventlog.write(source=cfg.notify.eventlog_source,
                           severity=Severity.INFO, event_id=520,
                           message=f"일일 알림 표시 상한 {cfg.notify.daily_alert_cap}건 도달. "
                                   f"남은 {len(others)}건은 표시하지 않음.")
            others = []

        # ---- 5) 표시 ------------------------------------------------------
        failures = 0
        # 5-a) CRITICAL / ERROR -> 강제 모달. 한 주기에 하나만 띄운다.
        modal_targets = criticals + [r for r in others if r.severity is Severity.ERROR]
        if modal_targets:
            top = modal_targets[0]
            if not _show_modal(conn, cfg, top):
                failures += 1
            # 모달은 사용자가 응답할 때까지 블록한다. 나머지는 다음 주기로 미룬다.
            return 1 if failures else 0

        # 5-b) WARN/INFO -> 토스트. 2건 이상이면 병합(L2).
        toast_targets = [r for r in others
                         if r.severity is Severity.WARN
                         or (r.severity is Severity.INFO and cfg.notify.show_info_toast)]
        if not toast_targets:
            return 0

        if toast.is_presentation_mode():
            eventlog.write(source=cfg.notify.eventlog_source,
                           severity=Severity.INFO, event_id=510,
                           message="전체 화면/프레젠테이션 모드 감지. "
                                   "토스트 표시를 다음 주기로 미룸.")
            return 0

        if len(toast_targets) >= cfg.notify.merge_threshold:
            if not _show_merged(conn, cfg, toast_targets):
                failures += 1
        else:
            for row in toast_targets[: cfg.notify.max_toasts_per_hour]:
                if not _show_toast(conn, cfg, row):
                    failures += 1

        mirror_to_file(conn, paths.ALERTS_MIRROR)
        return 1 if failures else 0
    finally:
        conn.close()


# ------------------------------------------------------------------ 표시부
def _show_toast(conn, cfg: Config, row: AlertRow) -> bool:
    buttons = tuple(
        toast.ToastButton(key, steps.label_of(key))
        for key in row.next_actions[:3]          # 토스트에는 최대 3개
    )

    def _on_click(key: str) -> None:
        if key == "snooze":
            snooze(conn, row.alert_id, minutes=cfg.notify.snooze_minutes)
            return
        steps.invoke(key, cfg, row.as_context())

    result = toast.show(
        title=row.title, body=row.body, severity=row.severity.value,
        buttons=buttons, seconds=cfg.notify.toast_seconds, on_click=_on_click,
    )

    if result.shown:
        mark_shown(conn, row.alert_id, channel="toast")
        eventlog.write_alert(cfg, row.code, row.severity, row.title, row.what,
                             event_id=eventlog.EVENT_IDS["NOTIFY_SHOWN"])
        return True

    attempts = mark_show_failed(conn, row.alert_id)
    eventlog.write(source=cfg.notify.eventlog_source, severity=Severity.WARN,
                   event_id=510,
                   message=f"[{row.code}] 토스트 표시 실패 ({attempts}회째).")
    if attempts >= 3:
        # R-P3: 3회 실패하면 모달 경로로 승격한다.
        return _show_modal(conn, cfg, row)
    return False


def _show_merged(conn, cfg: Config, rows: list[AlertRow]) -> bool:
    lines = [f"[{'주의' if r.severity is Severity.WARN else '정보'}] {r.title}"
             for r in rows[:5]]
    if len(rows) > 5:
        lines.append(f"그 외 {len(rows) - 5}건")

    def _on_click(key: str) -> None:
        steps.invoke(key, cfg, rows[0].as_context())

    result = toast.show_merged(
        count=len(rows), lines=lines, severity="WARN",
        seconds=cfg.notify.toast_seconds, on_click=_on_click,
    )
    if not result.shown:
        for row in rows:
            mark_show_failed(conn, row.alert_id)
        return False
    for row in rows:
        mark_shown(conn, row.alert_id, channel="toast")
        eventlog.write_alert(cfg, row.code, row.severity, row.title, row.what,
                             event_id=eventlog.EVENT_IDS["NOTIFY_SHOWN"])
    return True


def _show_modal(conn, cfg: Config, row: AlertRow) -> bool:
    # CRITICAL 은 화면 표시 성공 여부와 무관하게 웹훅을 병렬 발사한다.
    if row.severity is Severity.CRITICAL:
        webhook.send_alert(cfg, row)

    eventlog.write_alert(cfg, row.code, row.severity, row.title, row.body,
                         event_id=_eventlog_id_for(row))

    focus_key = _CHECK_KEY_BY_CODE.get(row.code)
    try:
        from dmf_crawler.gui import app as gui_app
        gui_app.launch(mode="recover", focus_key=focus_key, alert_id=row.alert_id)
    except Exception:
        eventlog.write(source=cfg.notify.eventlog_source, severity=Severity.ERROR,
                       event_id=510,
                       message=f"[{row.code}] 복구 창을 띄우지 못했습니다. "
                               f"MessageBox 폴백으로 강등합니다. "
                               + traceback.format_exc(limit=2))
        return _show_messagebox_fallback(conn, cfg, row)

    mark_shown(conn, row.alert_id, channel="modal")
    return True


def _show_messagebox_fallback(conn, cfg: Config, row: AlertRow) -> bool:
    """tkinter 가 완전히 불가능할 때 PowerShell 로 MessageBox 를 띄운다."""
    import subprocess
    script = paths.SCRIPTS_DIR / "notify.ps1"
    try:
        subprocess.run(
            ["powershell.exe", "-NoProfile", "-NonInteractive",
             "-WindowStyle", "Hidden", "-ExecutionPolicy", "Bypass",
             "-File", str(script), "-Mode", "Force"],
            timeout=180, creationflags=steps.CREATE_NO_WINDOW,
        )
    except Exception:
        mark_show_failed(conn, row.alert_id)
        return False
    mark_shown(conn, row.alert_id, channel="messagebox")
    return True


def _eventlog_id_for(row: AlertRow) -> int:
    from dmf_crawler.notify import messages
    tpl = messages.TEMPLATES.get(row.code)
    if tpl and tpl.eventlog_id:
        return tpl.eventlog_id
    return {"WARN": 200, "ERROR": 300, "CRITICAL": 400}.get(row.severity.value, 200)


# 알림 코드 -> 복구 GUI 에서 포커스할 체크 항목 키
_CHECK_KEY_BY_CODE: dict[str, str] = {
    "API_KEY_MISSING":  "api_key",
    "API_KEY_INVALID":  "api_key_live",
    "AGY_MISSING":      "agy_binary",
    "AGY_AUTH":         "agy_auth",
    "DB_CORRUPT":       "db_integrity",
    "MIGRATION_FAILED": "db_schema",
    "TASK_MISSING":     "tasks",
    "DISK_LOW":         "disk_free",
    "REPORT_FAILED":    "report_writable",
    "REPORT_LOCKED":    "report_writable",
    "WATCHDOG_STALE":   "last_run",
    "CONSECUTIVE_FAILURES": "last_run",
    "RUN_PAUSED":       "last_run",
}


def _auto_resolve(conn, cfg: Config) -> None:
    """상태성 CRITICAL 은 해당 체크가 통과하면 자동 해소한다(R-D2)."""
    from dmf_crawler.alerts import resolve
    open_codes = {r.code for r in pending(conn, modal_repeat_minutes=10 ** 6)}
    for code, check_key in _CHECK_KEY_BY_CODE.items():
        if code not in open_codes:
            continue
        try:
            outcome = checks.run_one(check_key, cfg)
        except Exception:
            continue
        if outcome.ok:
            n = resolve(conn, code, note=f"진단 '{check_key}' 통과로 자동 해소")
            if n:
                eventlog.write(
                    source=cfg.notify.eventlog_source, severity=Severity.INFO,
                    event_id=500,
                    message=f"[{code}] 문제가 해결돼 알림을 지웠습니다.")


def main(argv: list[str] | None = None) -> int:
    """cli.py 의 notify-pump 서브커맨드 본체."""
    try:
        cfg = load_config()
    except Exception:
        # 설정조차 못 읽으면 이벤트 로그가 유일한 통로다.
        eventlog.write(source="DMF Crawler", severity=Severity.CRITICAL,
                       event_id=900,
                       message="설정을 읽지 못해 알림 에이전트를 시작하지 못했습니다: "
                               + traceback.format_exc(limit=3))
        return 2
    try:
        return pump_once(cfg)
    except Exception:
        eventlog.write(source=cfg.notify.eventlog_source, severity=Severity.CRITICAL,
                       event_id=900,
                       message="알림 에이전트가 예외로 종료됐습니다: "
                               + traceback.format_exc(limit=5))
        return 2


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

6.7 `src/dmf_crawler/notify/webhook.py` 전문

# src/dmf_crawler/notify/webhook.py
"""웹훅 알림과 dead-man switch. 선택 기능(기본 off).

원칙:
    - 절대 예외를 밖으로 던지지 않는다.
    - 재시도하지 않는다(1회만). 웹훅 실패는 화면 알림을 대체하지 않는다.
    - URL 은 DPAPI 로 암호화 저장된다. config.toml 에 평문으로 두지 않는다.
    - 본문에 인증키·토큰을 절대 포함하지 않는다(mask 적용).
"""
from __future__ import annotations

import json
import re
from typing import Any

import httpx

from dmf_crawler import secrets_dpapi
from dmf_crawler.alerts import AlertRow, Severity
from dmf_crawler.config import Config

_SECRET_NAME = "webhook_url"
_DEADMAN_NAME = "deadman_url"

# 로그·웹훅 본문에서 지워야 하는 패턴
_MASK_PATTERNS = (
    re.compile(r"(serviceKey=)[^&\s]+", re.I),
    re.compile(r"(access_token[\"'\s:=]+)[A-Za-z0-9._\-]+", re.I),
    re.compile(r"\bya29\.[A-Za-z0-9._\-]+"),
)


def _mask(text: str) -> str:
    out = text
    for pat in _MASK_PATTERNS:
        out = pat.sub(lambda m: (m.group(1) if m.lastindex else "") + "***", out)
    return out


def _url() -> str | None:
    try:
        value = secrets_dpapi.load(_SECRET_NAME)
    except Exception:
        return None
    return value.strip() if value else None


def _payload(cfg: Config, title: str, body: str, severity: Severity) -> dict[str, Any]:
    kind = cfg.notify.webhook_kind
    icon = {"INFO": "ℹ️", "WARN": "⚠️", "ERROR": "⛔", "CRITICAL": "🚨"}.get(
        severity.value, "⚠️")
    text = f"{icon} **{title}**\n```\n{body}\n```"

    if kind == "discord":
        return {"content": text[:1900]}
    if kind == "slack":
        return {"text": text[:3000]}
    if kind == "telegram":
        # 텔레그램은 URL 에 chat_id 가 포함된 형태를 전제로 한다.
        return {"text": text[:4000], "parse_mode": "Markdown"}
    return {"title": title, "body": body, "severity": severity.value}


def send(cfg: Config, *, title: str, body: str, severity: Severity) -> bool:
    """웹훅 1건 발사. 성공하면 True."""
    if not cfg.notify.webhook_enabled:
        return False
    if not severity.at_least(Severity(cfg.notify.webhook_min_severity)):
        return False
    url = _url()
    if not url:
        return False

    payload = _payload(cfg, _mask(title), _mask(body), severity)
    try:
        with httpx.Client(timeout=cfg.notify.webhook_timeout_seconds) as client:
            resp = client.post(url, json=payload)
        return 200 <= resp.status_code < 300
    except Exception:
        return False


def send_alert(cfg: Config, row: AlertRow) -> bool:
    body = (f"{row.body}\n\n"
            f"발생: {row.last_seen_at}  (누적 {row.occurrences}회)\n"
            f"코드: {row.code}")
    return send(cfg, title=row.title, body=body, severity=row.severity)


# ------------------------------------------------------- dead-man switch
def ping(cfg: Config, event: str) -> bool:
    """healthchecks.io 계열 dead-man switch.

    event: "start" | "success" | "fail"
    PC 가 통째로 꺼져 있어도 상대편이 알아채는 유일한 경로다.
    """
    if not cfg.notify.deadman_enabled:
        return False
    try:
        base = secrets_dpapi.load(_DEADMAN_NAME)
    except Exception:
        return False
    if not base:
        return False
    base = base.rstrip("/")
    suffix = {"start": "/start", "success": "", "fail": "/fail"}.get(event, "")
    try:
        with httpx.Client(timeout=cfg.notify.webhook_timeout_seconds) as client:
            resp = client.get(base + suffix)
        return 200 <= resp.status_code < 300
    except Exception:
        return False


def test(cfg: Config) -> tuple[bool, str]:
    """온보딩 GUI 의 '웹훅 테스트' 버튼이 부르는 함수."""
    url = _url()
    if not url:
        return False, "웹훅 주소가 저장돼 있지 않습니다."
    ok = send(cfg,
              title="DMF 크롤러 — 웹훅 연결 시험",
              body="이 메시지가 보이면 웹훅이 정상 연결됐습니다.\n"
                   "앞으로 심각한 문제가 생기면 여기로 알려드립니다.",
              severity=Severity.CRITICAL)
    return (True, "테스트 메시지를 보냈습니다. 채널을 확인하세요.") if ok \
        else (False, "전송에 실패했습니다. 주소와 인터넷 연결을 확인하세요.")

6.8 설정 키 증분 (AMD-04)

아키텍처 §6.1 의 [notify] 섹션에 아래 키를 추가한다. 기존 5개는 그대로 유지된다.

키	타입	기본값	설명
`notify.modal_repeat_minutes`	int	`60`	ERROR·CRITICAL 모달 재촉 주기
`notify.snooze_minutes`	int	`60`	"나중에" 를 눌렀을 때 조용해지는 시간
`notify.merge_threshold`	int	`2`	이 건수 이상이면 토스트를 한 장으로 병합
`notify.max_toasts_per_hour`	int	`6`	시간당 토스트 상한
`notify.daily_alert_cap`	int	`20`	하루 알림 표시 상한(CRITICAL 면제)
`notify.show_info_toast`	bool	`false`	INFO 등급도 토스트로 띄울지
`notify.pump_stamp_stale_minutes`	int	`45`	이보다 오래되면 `notify.ps1` 폴백 발동
`notify.escalation_days`	int	`3`	연속 실패 이 일수부터 CRITICAL
`notify.escalation_webhook_days`	int	`5`	이 일수부터 웹훅 강제 발사
`notify.escalation_stop_after_days`	int	`7`	이 일수부터 자동 실행 일시중지
`notify.webhook_enabled`	bool	`false`	웹훅 사용
`notify.webhook_kind`	str	`"discord"`	`discord` \| `slack` \| `telegram` \| `generic`
`notify.webhook_min_severity`	str	`"CRITICAL"`	이 등급 이상만 웹훅 발사
`notify.webhook_timeout_seconds`	float	`10.0`	웹훅 타임아웃
`notify.deadman_enabled`	bool	`false`	dead-man switch 사용

config/config.toml 의 [notify] 완성형

[notify]
# --- 감지 ---------------------------------------------------------------
watchdog_stale_minutes        = 120     # heartbeat 신선도 한계(06:00 + 2시간)
pump_stamp_stale_minutes      = 45      # 이보다 오래되면 PowerShell 폴백 발동
consecutive_failure_critical  = 3       # 같은 WARN 이 이 횟수 반복되면 CRITICAL 승격

# --- 표시 ---------------------------------------------------------------
toast_seconds                 = 12      # 자동 소멸 알림 표시 시간
modal_repeat_minutes          = 60      # ERROR/CRITICAL 재촉 주기
snooze_minutes                = 60      # "나중에" 를 눌렀을 때 조용해지는 시간
show_info_toast               = false   # INFO 도 토스트로 띄울지

# --- 폭주 억제 -----------------------------------------------------------
cooldown_minutes              = 240     # 동일 코드 알림 억제 시간(요구 R7.8)
merge_threshold               = 2       # 이 건수 이상이면 한 장으로 병합
max_toasts_per_hour           = 6
daily_alert_cap               = 20      # CRITICAL 은 이 상한을 적용받지 않는다

# --- 에스컬레이션 ---------------------------------------------------------
escalation_days               = 3       # 연속 실패 3일 -> CRITICAL + 진단 자동 표시
escalation_webhook_days       = 5       # 5일 -> 웹훅 강제 발사
escalation_stop_after_days    = 7       # 7일 -> 자동 실행 일시중지

# --- 기록 ---------------------------------------------------------------
eventlog_source               = "DMF Crawler"

# --- 보조 채널(선택, 기본 꺼짐) --------------------------------------------
# 주소는 config 에 넣지 않는다. 온보딩 GUI 에서 입력하면 DPAPI 로 암호화 저장된다.
webhook_enabled               = false
webhook_kind                  = "discord"   # discord | slack | telegram | generic
webhook_min_severity          = "CRITICAL"
webhook_timeout_seconds       = 10.0
deadman_enabled               = false

7. 알림 테스트 절차

7.1 `dmf alert-test` 서브커맨드 (AMD-05)

아키텍처 §5 의 서브커맨드 표에 alert-test 를 추가한다. 근거: 알림 계층은 "실패해야만 검증되는" 코드다. 실제 장애를 기다려 검증하면 영원히 검증되지 않는다. 인위적 유발 수단이 없으면 §7.2 의 절차 절반이 실행 불가능하다.

커맨드	인자	동작
`alert-test`	`--code <ALERT_CODE>`	해당 코드의 알림을 더미 컨텍스트로 1건 발생시킨다(DB 에 기록)
	`--all`	등록된 모든 템플릿을 순회하며 문구를 렌더링만 하고 콘솔에 출력(발생시키지 않음)
	`--show`	발생 후 즉시 `pump_once()` 를 불러 화면 표시까지 확인
	`--channel {toast,modal,messagebox,msgexe,webhook,eventlog}`	특정 채널만 강제로 시험
	`--cleanup`	`alert-test` 가 만든 알림·이벤트를 전부 해소·삭제

# src/dmf_crawler/cli.py (발췌)
def cmd_alert_test(args) -> int:
    from dmf_crawler.config import load_config
    from dmf_crawler.notify import messages, eventlog, toast, webhook
    from dmf_crawler.storage import db
    from dmf_crawler import alerts, paths
    from dmf_crawler.notify import pump

    cfg = load_config()

    # --all : 전 템플릿 렌더링 검사. DB 를 건드리지 않는다.
    if args.all:
        for code, tpl in sorted(messages.TEMPLATES.items()):
            rendered = tpl.render(_DUMMY_CTX)
            print("=" * 72)
            print(f"[{tpl.severity.value}] {code}")
            print(f"제목: {rendered.title}")
            print(rendered.body)
            print("버튼: " + "  ".join(
                f"[{steps.label_of(k)}]" for k in rendered.next_actions))
        return 0

    # --channel : 채널 단독 시험. 알림을 만들지 않는다.
    if args.channel:
        return _test_channel(cfg, args.channel)

    code = args.code
    if code not in messages.TEMPLATES:
        print(f"알 수 없는 코드: {code}")
        print("사용 가능: " + ", ".join(sorted(messages.TEMPLATES)))
        return 2

    tpl = messages.TEMPLATES[code]
    rendered = tpl.render(_DUMMY_CTX)
    conn = db.connect(cfg, readonly=False)
    try:
        if args.cleanup:
            n = 0
            for c in messages.TEMPLATES:
                n += alerts.resolve(conn, c, note="alert-test --cleanup")
            print(f"{n}건을 해소했습니다.")
            return 0

        fired = alerts.raise_alert(
            conn, run_id="TEST", severity=tpl.severity, code=code,
            what=rendered.what, why=rendered.why, how=rendered.how,
            next_actions=rendered.next_actions,
            context=dict(_DUMMY_CTX), log_dir=paths.LOGS_DIR,
            cooldown_minutes=0,               # 시험에서는 쿨다운을 무시한다
            date_scoped=tpl.date_scoped, source="alert-test",
        )
        alerts.mirror_to_file(conn, paths.ALERTS_MIRROR)
        print(f"{code} 알림을 기록했습니다 (표시대상={fired}).")
        if args.show:
            rc = pump.pump_once(cfg)
            print(f"pump_once 종료 코드: {rc}")
        return 0
    finally:
        conn.close()


_DUMMY_CTX: dict[str, object] = {
    "run_date": "2026-09-02", "run_time": "06:04", "check_time": "2026-09-02 06:04",
    "now_time": "08:15", "stale_date": "2026-09-01",
    "pages_ok": 7, "pages_total": 12, "attempts": 4,
    "last_error_short": "연결 시간 초과 (30초)",
    "fail_count": 3, "cooldown_hours": 24, "resume_at": "2026-09-03 06:00",
    "page_no": 3, "body_bytes": 84, "min_bytes": 200,
    "total_count": 15832, "prev_count": 15840, "curr_count": 14851,
    "drop_count": 989, "drop_pct": "6.2", "threshold_pct": "5.0",
    "gate_name": "수집 건수 일치 검사", "gate_detail": "받은 건수가 API 가 알려준 전체 건수와 다릅니다.",
    "gate_threshold": "0건 차이", "gate_observed": "12건 차이",
    "missing_fields_str": "등록번호, 제조원", "null_ratio_pct": "37.4",
    "baseline_null_pct": "1.0", "proposal_path": r"D:\workspace\DMF_Crawler\state\proposals\2026-09-02.json",
    "result_code": "30", "result_msg": "SERVICE KEY IS NOT REGISTERED ERROR",
    "key_fingerprint": "9f3aB1c2", "error_code": "LIMITED_NUMBER_OF_SERVICE_REQUESTS_EXCEEDS_ERROR",
    "calls_today": 10021,
    "agy_path": r"C:\Users\encep\AppData\Local\agy\bin\agy.exe",
    "agy_error_short": "authentication required", "occurrences": 2,
    "tokens_used": 298_500, "tokens_cap": 300_000, "timeout": "10m",
    "fallback_name": "DMF_리포트_2026-09-02_060412.xlsx",
    "target_name": "DMF_리포트_2026-09-02.xlsx", "retries": 3,
    "report_dir": r"D:\workspace\DMF_Crawler\reports",
    "report_path": r"D:\workspace\DMF_Crawler\reports\DMF_리포트_2026-09-02.xlsx",
    "failure_summary": "시트 '성분별 집계' 생성 중 오류가 발생했습니다.",
    "drive": "D:", "free_gb": "1.4", "min_gb": "2.0", "reclaim_mb": 830,
    "busy_timeout_s": 15, "db_path": r"D:\workspace\DMF_Crawler\data\dmf.sqlite3",
    "integrity_result": "*** in database main *** Page 412: btreeInitPage() returns error code 11",
    "backup_count": 12, "latest_backup_date": "2026-09-01",
    "backup_dir": r"D:\workspace\DMF_Crawler\backup",
    "from_version": 2, "to_version": 3,
    "pre_migration_backup": r"D:\workspace\DMF_Crawler\backup\dmf_premigrate_0003.sqlite3",
    "last_success_at": "2026-09-01 06:03", "stale_hours": 26,
    "exec_limit_min": 30, "missing_tasks_str": "DMF_Crawler_Daily",
    "failed_days": 3, "first_failed_date": "2026-08-31", "last_failed_date": "2026-09-02",
    "top_cause_1": "AGY_AUTH", "top_cause_1_count": 3,
    "top_cause_2": "FETCH_FAILED", "top_cause_2_count": 2,
    "top_cause_3": "INTEGRITY_BLOCKED", "top_cause_3_count": 1,
    "pause_day": 7, "pause_at": "2026-09-06 08:15",
    "pause_flag_path": r"D:\workspace\DMF_Crawler\state\paused.flag",
    "from_channel": "토스트", "to_channel": "복구 창", "reason": "창을 만들지 못했습니다",
    "cap": 20, "pending_n": 4, "stale_min": 47,
    "last_pump_at": "2026-09-02 07:30", "project_root": r"D:\workspace\DMF_Crawler",
    "new_n": 12, "chg_n": 3, "wdr_n": 1,
    "log_dir": r"D:\workspace\DMF_Crawler\logs\run_20260902_060012",
    "raw_dir": r"D:\workspace\DMF_Crawler\data\raw\2026-09-02",
}


def _test_channel(cfg, channel: str) -> int:
    from dmf_crawler.notify import eventlog, toast, webhook
    from dmf_crawler.alerts import Severity
    import subprocess
    from dmf_crawler import paths

    if channel == "toast":
        r = toast.show(
            title="DMF 크롤러 — 알림 시험",
            body="[무엇]   알림 채널 점검용 시험 메시지입니다.\n"
                 "[왜]     dmf alert-test --channel toast 로 직접 실행했습니다.\n"
                 "[어떻게] 이 창이 보이면 토스트 채널이 정상입니다.",
            severity="WARN",
            buttons=(toast.ToastButton("open_doctor", "자세히 보기"),
                     toast.ToastButton("dismiss", "닫기")),
            seconds=cfg.notify.toast_seconds,
        )
        print(f"표시됨={r.shown}, 눌린버튼={r.clicked}")
        return 0 if r.shown else 1

    if channel == "modal":
        from dmf_crawler.gui import app as gui_app
        return gui_app.launch(mode="inspect")

    if channel in ("messagebox", "msgexe"):
        rc = subprocess.run(
            ["powershell.exe", "-NoProfile", "-ExecutionPolicy", "Bypass",
             "-File", str(paths.SCRIPTS_DIR / "notify.ps1"), "-Mode", "Test"]
        ).returncode
        return rc

    if channel == "webhook":
        ok, msg = webhook.test(cfg)
        print(msg)
        return 0 if ok else 1

    if channel == "eventlog":
        ok = eventlog.write(source=cfg.notify.eventlog_source,
                            severity=Severity.WARN, event_id=500,
                            message="알림 채널 점검용 시험 기록입니다.")
        print("기록 성공" if ok else "기록 실패")
        return 0 if ok else 1

    print(f"알 수 없는 채널: {channel}")
    return 2

7.2 시나리오별 인위적 유발 절차

각 행은 "이렇게 하면 반드시 그 알림이 뜬다" 는 재현 절차다. 검증 담당자는 이 표를 그대로 따라가면 된다.

#	코드	유발 방법	기대 결과	원복
T01	`FETCH_FAILED`	`config.local.toml` 에 `[source] base_url = "https://127.0.0.1:9/none"` 지정 후 `dmf run --force`	4회 재시도 후 WARN 토스트. 리포트는 스테일 자료로 생성	해당 줄 삭제
T02	`SOURCE_CIRCUIT_OPEN`	T01 을 3회 연속 실행	3회차에 CRITICAL 모달 1회. 4회차에는 침묵(전환 시에만 알림)	`DELETE FROM component_health WHERE component='source_mfds'`
T03	`HTTP_BLOCKED_BODY`	로컬에 200 + 본문 `<html>점검중</html>` 을 반환하는 1줄 서버를 띄우고 `base_url` 을 그리로	WARN 토스트. `data/raw/` 에 원문 보존	설정 원복
T04	`ZERO_RECORDS`	로컬 서버가 `{"response":{"body":{"totalCount":0,"items":[]}}}` 반환	CRITICAL 모달 + 웹훅(켜져 있으면). 스냅샷 미저장	설정 원복
T05	`INTEGRITY_BLOCKED`	`config.local.toml` 에 `[integrity] max_null_ratio = 0.0`	WARN 토스트. diff 미수행	값 원복
T06	`INTEGRITY_DROP`	`[integrity] max_drop_ratio = 0.0000001` 로 낮춤	CRITICAL 모달	값 원복
T07	`SCHEMA_DRIFT`	로컬 서버가 필드명을 바꾼 응답 반환(`dmfRegNo` → `regNo`)	CRITICAL 모달. 원문 보존	설정 원복
T08	`API_KEY_MISSING`	`dmf secrets delete service_key` 후 `dmf run`	종료 코드 2, CRITICAL 강제 모달. GUI 에 키 입력 버튼	키 재입력
T09	`API_KEY_INVALID`	인증키를 `INVALID_TEST_KEY` 로 저장 후 `dmf run --force`	CRITICAL 강제 모달. 지문 앞 8자만 표시되는지 확인	정상 키 복원
T10	`API_QUOTA_EXCEEDED`	로컬 서버가 `LIMITED_NUMBER_OF_SERVICE_REQUESTS_EXCEEDS_ERROR` 반환	WARN 토스트. 리포트 생성됨	설정 원복
T11	`AGY_MISSING`	`config.local.toml` 에 `[agy] binary_path = "C:\\nope\\agy.exe"`	CRITICAL 모달. 리포트는 정상 생성(핵심 확인 항목)	값 삭제
T12	`AGY_AUTH`	`~/.gemini/antigravity-cli/antigravity-oauth-token` 을 `.bak` 로 이름 변경	CRITICAL 모달. [로그인 창 열기] 클릭 시 보이는 콘솔 창이 뜨는지 확인	파일 이름 복원
T13	`AGY_QUOTA`	`[agy] daily_token_cap = 1` 로 설정 후 `dmf run --force`	WARN 토스트. 대시보드 시트에 "AI 요약 없음" 배지	값 원복
T14	`REPORT_LOCKED`	오늘 리포트를 Excel 로 열어 둔 채 `dmf run --force`	WARN 토스트 + `_HHMMSS` 폴백 파일 생성	Excel 닫고 `dmf report-only`
T15	`REPORT_FAILED`	`reports/` 디렉터리를 읽기 전용으로 만들거나 이름을 바꿔 둠	ERROR 모달. [리포트 다시 만들기] 버튼 존재	권한 원복
T16	`DISK_LOW`	`[backup] min_free_gb = 99999`	WARN 토스트 + [정리하기] 버튼. 백업만 스킵	값 원복
T17	`DB_LOCKED`	DB Browser for SQLite 로 `dmf.sqlite3` 를 열고 쓰기 트랜잭션 시작 후 `dmf run --force`	ERROR 모달	도구 닫기
T18	`DB_CORRUPT`	DB 사본을 만들어 헥스 편집기로 중간 바이트를 훼손하고 `[storage] sqlite_path` 를 그리로	CRITICAL 강제 모달 + [백업으로 복원]	설정 원복
T19	`WATCHDOG_STALE`	`state/heartbeat.json` 의 `last_success_at` 을 3일 전으로 수정 후 `dmf notify-pump --once`	CRITICAL 강제 모달	파일 삭제 후 정상 실행
T20	`TASK_MISSING`	`Disable-ScheduledTask -TaskName DMF_Crawler_Daily`	CRITICAL 모달 + [자동 실행 다시 등록]	`Enable-ScheduledTask`
T21	`CONSECUTIVE_FAILURES`	`runs` 테이블에 최근 3일치 `status='FAILED'` 행을 직접 INSERT 후 pump 실행	CRITICAL 모달 + 원인 상위 3개 표시	해당 행 DELETE
T22	`RUN_PAUSED`	위와 같이 7일치 FAILED INSERT 후 pump 실행	`state/paused.flag` 생성 + CRITICAL 모달. 다음 `dmf run` 이 즉시 종료되는지 확인	플래그 삭제
T23	`PYTHON_BROKEN`	`.venv` 폴더를 `.venv_bak` 로 이름 변경 후 Agent 작업 수동 실행	pump 액션 실패 → `notify.ps1 -Mode Guard` 가 MessageBox 표시	이름 복원
T24	병합(L2)	`dmf alert-test --code FETCH_FAILED`, `--code REPORT_LOCKED`, `--code AGY_QUOTA` 를 연속 실행 후 `dmf notify-pump --once`	토스트 1장에 3건 요약	`--cleanup`
T25	상한(L3)	`[notify] daily_alert_cap = 1` 로 낮추고 알림 2건 발생	2번째는 표시되지 않고 이벤트 ID 520 기록	값 원복
T26	쿨다운(L1)	같은 코드로 `alert-test` 를 2회 연속 실행(쿨다운 240분 기본값 사용)	2번째는 `표시대상=False`, `occurrences=2`	`--cleanup`
T27	로그오프 축적(B1)	알림 발생 후 로그오프 → 재로그온	로그온 직후 `AtLogOn` 트리거로 밀린 알림 표시	—
T28	전체 화면(B4)	게임/PPT 를 전체 화면으로 띄운 상태에서 WARN 발생	토스트가 미뤄지고 이벤트 ID 510 기록. 창을 내리면 다음 주기에 표시	—

7.3 채널별 단독 점검 (설치 직후 필수)

# 1) 토스트 — 우하단에 12초짜리 창이 떠야 한다
D:\workspace\DMF_Crawler\.venv\Scripts\python.exe -m dmf_crawler alert-test --channel toast

# 2) 복구 GUI 모달 — 진단 체크리스트 창이 떠야 한다
D:\workspace\DMF_Crawler\.venv\Scripts\pythonw.exe -m dmf_crawler alert-test --channel modal

# 3) MessageBox 폴백 — 항상 맨 앞에 뜨는지 확인
powershell -NoProfile -ExecutionPolicy Bypass `
    -File D:\workspace\DMF_Crawler\scripts\notify.ps1 -Mode Test

# 4) 이벤트 로그
D:\workspace\DMF_Crawler\.venv\Scripts\python.exe -m dmf_crawler alert-test --channel eventlog
Get-WinEvent -FilterHashtable @{LogName='Application'; ProviderName='DMF Crawler'} -MaxEvents 3 |
    Format-List TimeCreated, Id, LevelDisplayName, Message

# 5) 웹훅(켜 두었을 때만)
D:\workspace\DMF_Crawler\.venv\Scripts\python.exe -m dmf_crawler alert-test --channel webhook

# 6) 전체 문구 렌더링 검사 — 치환 누락, 깨진 줄바꿈, 과도한 길이 확인
D:\workspace\DMF_Crawler\.venv\Scripts\python.exe -m dmf_crawler alert-test --all

7.4 자동 회귀 테스트

tests/test_alerts.py 로 CI 없이도 pytest 한 번에 검증되는 항목들이다.

# tests/test_alerts.py
import json
import pytest

from dmf_crawler.alerts import Severity, raise_alert, pending, resolve, snooze
from dmf_crawler.notify import messages


def test_모든_템플릿이_4요소를_갖는다():
    """messages 모듈 임포트만으로 계약이 검증된다(_validate_registry)."""
    assert messages.TEMPLATES                      # 임포트가 성공했다는 것이 곧 통과
    for code, tpl in messages.TEMPLATES.items():
        assert tpl.what and tpl.why and tpl.how, code
        assert tpl.next_actions, code


def test_실패_알림에_실행가능한_액션이_있다():
    for code, tpl in messages.TEMPLATES.items():
        if tpl.severity.at_least(Severity.WARN):
            actionable = set(tpl.next_actions) - {"snooze", "dismiss"}
            assert actionable, f"{code}: 막다른 골목 알림(R7.7 위반)"


def test_모든_액션키가_레지스트리에_존재한다():
    from dmf_crawler.gui import steps
    for code, tpl in messages.TEMPLATES.items():
        for key in tpl.next_actions:
            assert key in steps.ACTIONS, f"{code}: 미등록 액션 {key}"


def test_이벤트로그_ID가_1에서_1000_범위다():
    """eventcreate.exe 의 하드 제약."""
    from dmf_crawler.notify import eventlog
    for name, eid in eventlog.EVENT_IDS.items():
        assert 1 <= eid <= 1000, f"{name}={eid} 는 eventcreate 범위를 벗어난다"
    for code, tpl in messages.TEMPLATES.items():
        if tpl.eventlog_id:
            assert 1 <= tpl.eventlog_id <= 1000, code


def test_4요소_누락시_ValueError(tmp_conn):
    with pytest.raises(ValueError, match="why"):
        raise_alert(tmp_conn, run_id="T", severity=Severity.WARN, code="X",
                    what="무엇", why="   ", how="어떻게",
                    next_actions=("run_now",), cooldown_minutes=0)


def test_막다른골목_알림은_거부된다(tmp_conn):
    with pytest.raises(ValueError, match="R7.7"):
        raise_alert(tmp_conn, run_id="T", severity=Severity.CRITICAL, code="X",
                    what="a", why="b", how="c",
                    next_actions=("dismiss",), cooldown_minutes=0)


def test_쿨다운_안에서는_표시대상이_아니다(tmp_conn):
    kw = dict(run_id="T", severity=Severity.WARN, code="FETCH_FAILED",
              what="a", why="b", how="c", next_actions=("run_now",),
              cooldown_minutes=240)
    assert raise_alert(tmp_conn, **kw) is True
    assert raise_alert(tmp_conn, **kw) is False        # 쿨다운 안
    rows = pending(tmp_conn)
    assert len(rows) == 1
    assert rows[0].occurrences == 2                     # 카운터는 올라간다


def test_alert_events는_매번_쌓인다(tmp_conn):
    kw = dict(run_id="T", severity=Severity.WARN, code="FETCH_FAILED",
              what="a", why="b", how="c", next_actions=("run_now",),
              cooldown_minutes=240)
    raise_alert(tmp_conn, **kw)
    raise_alert(tmp_conn, **kw)
    n = tmp_conn.execute("SELECT COUNT(*) c FROM alert_events").fetchone()["c"]
    assert n == 2                                       # append-only 불변식


def test_해소된_알림은_pending에_없다(tmp_conn):
    raise_alert(tmp_conn, run_id="T", severity=Severity.CRITICAL, code="AGY_AUTH",
                what="a", why="b", how="c", next_actions=("agy_relogin",),
                cooldown_minutes=0, date_scoped=False)
    assert len(pending(tmp_conn)) == 1
    assert resolve(tmp_conn, "AGY_AUTH", note="테스트") == 1
    assert pending(tmp_conn) == []


def test_스누즈_동안은_표시되지_않는다(tmp_conn):
    raise_alert(tmp_conn, run_id="T", severity=Severity.CRITICAL, code="AGY_AUTH",
                what="a", why="b", how="c", next_actions=("agy_relogin",),
                cooldown_minutes=0, date_scoped=False)
    row = pending(tmp_conn)[0]
    snooze(tmp_conn, row.alert_id, minutes=60)
    assert pending(tmp_conn) == []


def test_문구에_비밀값_패턴이_없다():
    """문구 템플릿이 실수로 키·토큰을 노출하지 않는지."""
    banned = ("serviceKey=", "access_token", "ya29.")
    for code, tpl in messages.TEMPLATES.items():
        blob = " ".join((tpl.title, tpl.what, tpl.why, tpl.how))
        for word in banned:
            assert word not in blob, f"{code}: 비밀값 패턴 '{word}' 노출"


def test_제목이_60자를_넘지_않는다():
    for code, tpl in messages.TEMPLATES.items():
        assert len(tpl.title) <= 60, f"{code}: {len(tpl.title)}자"

7.5 설치 직후 승인 체크리스트

운영 담당자가 최초 설치 후 한 번 통과시키는 목록이다. 하나라도 실패하면 배치를 신뢰할 수 없다.

alert-test --all 이 전 템플릿을 오류 없이 렌더링하고, (정보 없음) 이 하나도 안 보인다
alert-test --channel toast — 우하단에 창이 뜨고 12초 뒤 사라진다
토스트에 마우스를 올리면 사라지지 않고, 떼면 3초 뒤 사라진다
alert-test --channel modal — 복구 창이 뜨고 진단 12항목이 보인다
notify.ps1 -Mode Test — MessageBox 가 다른 창들보다 앞에 뜬다
alert-test --channel eventlog 후 이벤트 뷰어에 "DMF Crawler" 원본이 보인다
alert-test --code AGY_AUTH --show — 모달이 뜨고 [로그인 창 열기]가 있다
그 버튼을 누르면 보이는 검은 콘솔 창이 뜨고 한국어 안내가 나온다
alert-test --code REPORT_LOCKED --show — 토스트 [리포트 다시 만들기]가 실제로 동작한다
T24(병합)를 수행하면 창이 3장이 아니라 1장 뜬다
T19(워치독)를 수행하면 CRITICAL 모달이 뜬다
T23(.venv 제거)을 수행하면 MessageBox 폴백이 뜬다 — 가장 중요한 항목
위 시험 후 alert-test --cleanup 으로 시험 알림이 전부 사라진다
state/alerts.json 의 pending 이 빈 배열이 된다

8. 에스컬레이션

8.1 3단 에스컬레이션

일차	트리거	자동 조치	사용자에게 보이는 것
1~2일	`runs.status='FAILED'`	없음. 다음 실행을 기다린다	매일 ERROR 모달 1회(60분마다 재촉)
3일	`consecutive_failed_days >= notify.escalation_days`	`CONSECUTIVE_FAILURES` CRITICAL 발생. 복구 GUI 가 `inspect` 모드로 자동 기동되어 진단 12종을 즉시 보여준다	원인 상위 3개가 적힌 CRITICAL 모달 + 진단 화면
5일	`>= notify.escalation_webhook_days`	웹훅 강제 발사. `webhook_min_severity` 설정과 무관하게, 그리고 `webhook_enabled=false` 여도 URL 이 저장돼 있으면 보낸다	디스코드/슬랙/텔레그램 메시지
7일	`>= notify.escalation_stop_after_days`	`state/paused.flag` 생성 → 다음 `dmf run` 이 즉시 종료 0. API 호출을 멈춘다	`RUN_PAUSED` CRITICAL 모달. [자동 실행 재개] 버튼은 진단 전부 통과해야 활성화

8.2 왜 7일에 멈추는가

근거	설명
API 예의	고장난 상태로 매일 수십~수백 회 호출하면 공공데이터포털 쪽에 무의미한 부하를 준다. 요구 N1(정중한 접근)의 직접 귀결이다
쿼터 보호	개발계정 일일 10,000회 한도를 실패 재시도로 태우면, 정작 고친 날 쓸 수 없다
알림 피로 차단	7일째면 사용자는 이미 알림을 무시하고 있다. 조용해지되 멈췄다는 사실 자체를 알리는 것이 더 강한 신호다
되돌리기 쉬움	작업 스케줄러를 지우지 않고 플래그 파일 하나로 멈춘다. 재개는 파일 삭제 한 번이다

8.3 일시중지의 구현

# src/dmf_crawler/pipeline.py 의 stage_preflight 안 (발췌)
def _check_paused(ctx: RunContext) -> None:
    """일시중지 플래그가 있으면 즉시 SKIPPED 로 끝낸다.

    --force 로도 뚫리지 않는다. 사람이 명시적으로 재개해야 한다.
    (--force 는 '오늘 이미 성공했지만 다시 돌린다'는 뜻이지
     '고장난 채로 계속 두드린다'는 뜻이 아니다.)
    """
    if not paths.PAUSE_FLAG.exists():
        return
    try:
        info = json.loads(paths.PAUSE_FLAG.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        info = {}
    raise PausedError(
        f"연속 실패로 자동 실행이 멈춰 있습니다 "
        f"(멈춘 시각: {info.get('paused_at', '알 수 없음')}, "
        f"사유: {info.get('reason', '알 수 없음')}). "
        f"복구 창의 [자동 실행 재개]를 눌러 다시 켜세요."
    )

PausedError 는 run_pipeline 최상위에서 잡혀 status='SKIPPED', 종료 코드 0 으로 처리된다. 종료 코드 1 을 쓰면 Task Scheduler 가 RestartCount 재시도를 3번 더 돌려 "멈췄는데 계속 도는" 모순이 생긴다.

8.4 재개 경로

경로	조건	방법
GUI	진단 12종 전부 통과	복구 창 → [자동 실행 재개]
CLI	없음(강제)	`dmf run --resume` 또는 `del state\paused.flag`
자동	없다	자동 재개는 하지 않는다. 자동으로 멈춘 것을 자동으로 풀면 멈춘 의미가 사라진다

8.5 에스컬레이션 흐름도

매일 06:00 실행
   |
   +- SUCCESS -> heartbeat 갱신, 일자성 WARN 자동 해소(R-D1), 카운터 0 으로
   |
   +- FAILED
        |
        v
   Agent(15분 주기) 워치독이 consecutive_failed_days 계산
        |
   +----+-----------------------------------------+
   | 1~2일  : ERROR/CRITICAL 모달 (60분 재촉)      |
   | 3일    : CONSECUTIVE_FAILURES + 진단 자동 표시 |
   | 5일    : 웹훅 강제 발사                        |
   | 7일    : paused.flag 생성 + RUN_PAUSED         |
   +----+-----------------------------------------+
        |
        v
   사람이 원인 해결 -> 진단 통과 -> [자동 실행 재개]
        |
        v
   plag 삭제, 다음 06:00 정상 복귀
   (다음 SUCCESS 에서 R-D1/R-D2 가 모든 알림을 해소)

9. 이 문서가 상위 정본에 요구하는 개정 목록

문서 사이의 어긋남을 방지하기 위해, 이 문서가 docs/design/01-architecture.md 에 대해 만든 개정을 한곳에 모은다. 아키텍처를 갱신할 때 이 표를 그대로 반영하면 된다.

ID	대상	개정 내용	근거
AMD-01	§3.14 `alerts.py`	`Severity` 를 3값 → 4값(`INFO/WARN/ERROR/CRITICAL`)으로 확장	"리포트 미생성"과 "사람 개입 필수"는 채널·재촉 주기·에스컬레이션 카운터가 전부 다르다(§2.1)
AMD-02	§2 디렉터리 트리	`src/dmf_crawler/notify/webhook.py`, `scripts/notify.ps1`, `scripts/register_protocol.ps1` 3개 파일 추가	실패 시나리오 #32 에서 tkinter 는 원리적으로 뜰 수 없다. PowerShell 폴백이 유일한 화면 경로다(§6.0)
AMD-03	§3.9 불변식	`alerts` 를 `alert_events`(append-only 사실) + `alerts`(상태 머신)로 2분할	기존 불변식 "alerts 를 UPDATE 하지 않는다"를 깨지 않으면서 dedup·표시·해소 상태를 관리하는 유일한 방법(§2.3)
AMD-04	§6.1 설정 키	`[notify]` 에 15개 키 추가	폭주 억제 3층·에스컬레이션 3단·웹훅 채널에 필요(§6.8)
AMD-05	§5 서브커맨드 표	`alert-test` 추가	알림 계층은 인위적 유발 수단 없이는 검증 불가(§7.1)
AMD-06	§3.16 `pump.py`	`pump_once` 가 매 주기 `state/pump.stamp` 를 갱신하고, Agent 작업에 두 번째 액션(`notify.ps1 -Mode Guard`)을 등록	pump 가 죽었다는 사실 자체를 감지할 주체가 필요하다(§6.1)
AMD-07	§5 종료 코드	`PausedError` → `status='SKIPPED'` + 종료 코드 0	코드 1 이면 Task Scheduler 가 3회 재시도해 "멈췄는데 계속 도는" 모순이 생긴다(§8.3)
AMD-08	§7 실패 시나리오 대응표	시나리오 #28(배치 미실행)의 알림 코드를 `WATCHDOG_STALE` 로, #30(연속 3일)을 `CONSECUTIVE_FAILURES` 로 명시. 신규 #34 `RUN_PAUSED` 추가	코드 이름이 dedup_key 의 일부이므로 표기가 일치해야 한다

연구 문서 08 과의 차이 (08 은 리서치, 01-architecture 가 정본이다)

항목	연구 08	이 문서(=아키텍처 정본)	이유
작업 3종 이름	Daily / Watchdog / Notify	Daily / Agent / AgyUpdate	워치독을 별도 작업으로 두지 않고 Agent 안에서 판정한다(아키텍처 부록). 대신 주간 `agy update` 작업이 필요해졌다
토스트 구현	BurntToast	tkinter 자체 창	ADR-12. 외부 모듈 의존은 순환 실패를 만든다
이벤트 ID	1000 / 1001	100~910 (§2.8)	`eventcreate.exe` 의 `/ID` 는 1~1000 만 허용한다. 1001 은 실행 자체가 실패한다
dead-man switch	필수	선택(기본 off)	외부 서비스 계정이 전제이므로 비개발자 온보딩에서 기본값으로 강제할 수 없다. 온보딩에서 권유만 한다
웹훅	3단 폴백의 3단	병렬 채널	화면 표시 성공 여부와 무관하게 CRITICAL 은 발사한다. "화면을 봤다"와 "PC 앞에 있었다"는 다른 사실이다(§1.3)

부록. 미해결 / 실측 필요

A. 실측이 필요한 항목

eventcreate.exe 의 /ID 상한이 정확히 1000 인가. 문서상 1~1000 이지만 Windows 11 26xxx 빌드에서 재확인 필요. 1001 을 넣었을 때의 정확한 오류 메시지도 기록할 것. (⚠️ 미검증)
/SO 로 지정한 원본 이름이 Application 로그에 자동 등록되는가. 그룹 정책으로 Application 로그 쓰기가 제한된 환경에서의 동작. 최초 1회 관리자 권한이 필요한지. (⚠️ 미검증)
msg.exe 가 Windows 11 Pro 26220 에 존재하는가. Home 에디션에는 없는 것으로 알려져 있으나 Pro 실측 필요. /TIME: 파라미터의 실제 동작도. (⚠️ 미검증)
SHQueryUserNotificationState 의 Windows 11 반환값. 특히 값 7(APP, 전체 화면 앱)이 실제로 반환되는지, 그리고 집중 지원(방해 금지) ON 일 때 값 6(QUIET_TIME)이 오는지. 우리 tkinter 창은 QUIET_TIME 에도 떠야 하므로 판정에서 6 을 제외한 것이 옳은지 확인. (⚠️ 미검증)
overrideredirect(True) + -topmost 창이 잠금 화면·UAC 프롬프트 위에 뜨는가. 뜨면 안 된다(보안). 실제로는 뜨지 않을 것으로 보이나 확인 필요. (⚠️ 미검증)
Task Scheduler 다중 액션이 정말 앞 액션의 종료 코드를 무시하고 순차 실행하는가. §6.1 폴백 설계의 전제다. notify.ps1 이 pump 실패와 무관하게 실행되는지 실측. (⚠️ 미검증)
CREATE_NEW_CONSOLE 로 띄운 agy 가 실제로 기본 브라우저 OAuth 를 여는가. agy SSOT 는 "기본 브라우저로 OAuth 로그인"이라고 적었지만 재로그인(만료 후) 경로도 같은지 실측. 다르다면 §5.2 의 안내 배너 문구를 고쳐야 한다. (⚠️ 미검증)
agy 인증 만료 시 error 필드의 정확한 문자열과 종료 코드. classify_error 의 AUTH 판정 규칙이 여기에 달려 있다. agy SSOT 부록 B 에도 같은 항목이 미결로 남아 있다. (⚠️ 미검증)
agy 쿼터 소진 시 error 문자열. QUOTA 판정 규칙. 위와 동일. (⚠️ 미검증)
DPAPI 로 저장한 웹훅 URL 을 S4U 세션에서 복호화할 수 있는가. 연구 08 은 S4U 가 "encrypted files 접근 불가"라고 경고한다. DPAPI 사용자 범위가 여기 해당하는지가 관건이다 — 해당한다면 웹훅을 배치가 아니라 Agent 만 발사할 수 있고, 설계를 바꿔야 한다. 이 문서에서 가장 위험한 미검증 항목이다. (⚠️ 미검증, 우선순위 최상)
file_lock(msvcrt) 을 pump 와 파이프라인이 서로 다른 파일로 잡을 때 충돌이 없는가. run.lock 과 pump.lock 분리가 의도대로 동작하는지. (⚠️ 미검증)

B. 설계 결정을 미룬 항목

GEMINI_API_KEY 모드로 전환해 agy 로그인 자체를 없앨 것인가. agy SSOT §4.1 에 따르면 modelProvider: "gemini" + 환경변수로 완전 비대화형이 된다. 그러면 AGY_AUTH 시나리오 자체가 사라진다. 다만 과금 체계가 달라지는 결정이므로 알림 설계가 임의로 정할 수 없다. 사용자 판단 필요.
액션 센터 잔류가 필요한가. tkinter 토스트는 놓치면 사라진다. 요구 N3(24시간 내 인지)는 만족하지만, "자리를 비운 사이 WARN 이 지나갔다"를 사용자가 불편해하면 win11toast 를 선택 의존성으로 켜는 안을 재검토한다. 그때 §4.4 의 프로토콜 등록이 필요해진다.
이메일 채널을 정말 안 만들 것인가. 웹훅이 없는 사용자에게는 PC 밖 채널이 dead-man switch 뿐이다. 필요해지면 notify/webhook.py 에 smtp kind 를 추가하는 형태로 확장한다(파일 신설 없이).
알림 이력 화면. 지금은 alert_events 에 쌓기만 하고 사람이 볼 화면이 없다. 리포트 xlsx 의 s99_meta 시트에 최근 14일 알림 요약을 넣을지, 복구 GUI 에 탭을 하나 더 둘지 미정. 리포트 명세(docs/design/03-xlsx-report-spec.md) 작성 시 결정한다.
INTEGRITY_BLOCKED 의 게이트별 문구 분리. 지금은 게이트 이름을 {gate_name} 으로 치환하는 단일 템플릿이다. 게이트 2/4/5 의 사용자 조치가 실제로 다르다면 코드를 INTEGRITY_COUNT_MISMATCH / INTEGRITY_NULL_RATIO / INTEGRITY_DUPLICATE 로 쪼개야 한다. 실운영 데이터를 보고 결정한다.
다국어. general.language 키가 있지만 문구는 한국어 상수로 하드코딩돼 있다. 영어가 필요해지면 TEMPLATES 를 언어별 딕셔너리로 감싼다. 지금은 필요 없다.

C. 운영 중 재확인할 임계값

notify.cooldown_minutes = 240 이 적절한가. 하루 1회 배치이므로 240분이면 사실상 "하루 1회"다. 너무 조용하면 120 으로 내린다.
notify.daily_alert_cap = 20 이 실제로 도달하는 날이 있는가. 도달한다면 병합(L2) 임계값을 낮추는 편이 낫다.
notify.watchdog_stale_minutes = 120 이 06:00 + 2시간이라는 전제와 맞는가. schedule.daily_time 을 바꾸면 이 값도 함께 조정해야 한다 — 두 값을 연동시킬지, 독립으로 둘지 정할 것.
notify.escalation_stop_after_days = 7 이 너무 늦은가. 실운영에서 3일이면 이미 방치 상태라면 5일로 당긴다.
notify.toast_seconds = 12 로 4요소 본문(약 6줄)을 다 읽을 수 있는가. 못 읽으면 15~20 으로 올린다.

230 KiB Raw Permalink Blame History Unescape Escape

장애 알림과 복구 안내 설계

0. 한눈에 보기

1. 알림 채널 확정

1.1 채널 목록과 채택 근거

1.2 토스트가 안 뜨는 조건과 대응

1.3 폴백 사다리 (표시 경로 결정)

1.4 웹훅 채널 정책

1.5 채널 라우팅 매트릭스

2. 알림 등급 체계

2.1 4등급 정의 — AMD-01 (아키텍처 §3.14 개정)

2.2 등급별 채널·표시 방식·빈도

2.3 dedup_key 와 쿨다운

2.4 알림 폭주 억제 3중 방어

2.5 승격 규칙과 해소 규칙

2.6 alerts / alert_events DDL

2.7 state/alerts.json 미러 스키마

2.8 Windows 이벤트 로그 ID 배정

3. 알림 문구 전집

3.1 4요소 계약

3.2 시나리오 카탈로그

3.3 문구 전문

S02 FETCH_FAILED — 크롤링(수집) 실패 · WARN

S03 SOURCE_CIRCUIT_OPEN — 소스 차단·연속 실패로 호출 중단 · CRITICAL

S04 HTTP_BLOCKED_BODY — 차단 감지(200인데 내용이 없음) · WARN

S05 ZERO_RECORDS — 0건 수집 · CRITICAL

S06 INTEGRITY_BLOCKED — 무결성 게이트 차단 · WARN

S07 INTEGRITY_DROP — 전체 건수 급감 · CRITICAL

S08 SCHEMA_DRIFT — API 응답 구조 변경(구 "셀렉터 깨짐" 등가) · CRITICAL

S09 API_KEY_MISSING — API 키 미설정 · CRITICAL (강제 창)

S10 API_KEY_INVALID — 인증키 무효·만료 · CRITICAL (강제 창)

S11 API_QUOTA_EXCEEDED — 일일 트래픽 초과 · WARN

S12 AGY_MISSING — agy 미설치 · CRITICAL

S13 AGY_AUTH — agy 인증 만료 · CRITICAL (강제 창)

S14 AGY_QUOTA — agy 쿼터 소진 · WARN

S16 REPORT_LOCKED — xlsx 파일 잠김 · WARN

S17 REPORT_FAILED — 리포트 생성 실패 · ERROR (모달)

S18 DISK_LOW — 디스크 부족 · WARN

S19 DB_LOCKED — 데이터베이스 잠김 · ERROR (모달)

S20 DB_CORRUPT — 데이터베이스 손상 · CRITICAL (강제 창)

S21 MIGRATION_FAILED — 스키마 마이그레이션 실패 · CRITICAL (강제 창)

S22 WATCHDOG_STALE — 배치 미실행 워치독 경보 · CRITICAL (강제 창)

S23 TASK_MISSING — 작업 스케줄러 등록 소실 · CRITICAL

S24 CONSECUTIVE_FAILURES — 연속 3일 실패 · CRITICAL (강제 창)

S25 PYTHON_BROKEN — 알림 계층 자체가 죽음 · CRITICAL (MessageBox 폴백)

S27 RUN_PAUSED — 자동 실행 일시중지 · CRITICAL (강제 창)