# -*- coding: utf-8 -*- """utils.source_attribution_guard — Stop hook 사후 거짓표현 차단 가드. task-2644 ANU_CALLBACK_COLLECTOR_CONTROL_PLANE (회장 verbatim 보강-1 + §15 fixture-8) spec: memory/specs/system_anu_callback_collector_control_plane_spec_260524.md spec sha256: b27da557d4245bce476cd63f4ab174aefc8a25d2da07ec2c8d2c83b01ee96153 회장 verbatim 박제 (PR #146 1C0F6F52 사건 2026-05-23 ~00:40 KST): schedule_history 사후 조회 결과를 "callback received / 도착 / 수신" 처럼 inbound 수신으로 흐려 표현한 사건 → Stop hook 으로 사전 차단. ANCHOR-9: "거짓말 패턴 (수신 vs 사후 조회 흐림) Stop hook 으로 사전 차단". ANCHOR-11: "source attribution enum 8 (★ CALLBACK_COLLECTOR_PROCESSED 가 주 경로 · received_inbound 흐림 차단)". source enum 8 (회장 verbatim 5 → 7 → 8): 1. RECEIVED_INBOUND_THIS_SESSION 2. LOG_LOOKUP_AFTER_CHAIR_QUESTION 3. LOG_LOOKUP_PROACTIVE 4. MEMORY_RECALL 5. INFERENCE_ONLY 6. CALLBACK_COLLECTOR_PROCESSED (★ task-2644 주 경로) 7. CALLBACK_LEDGER_RECONCILED (post-hoc reconciliation) 8. LOG_LOOKUP_OR_SCHEDULE_HISTORY_VERIFICATION (★ §15 1C0F6F52 박제) """ from __future__ import annotations import re from enum import Enum from typing import Dict, List, Optional, Tuple SCHEMA = "utils.source_attribution_guard.v1" class SourceAttribution(str, Enum): """회장 verbatim 8 enum (보강-1 + §15).""" RECEIVED_INBOUND_THIS_SESSION = "RECEIVED_INBOUND_THIS_SESSION" LOG_LOOKUP_AFTER_CHAIR_QUESTION = "LOG_LOOKUP_AFTER_CHAIR_QUESTION" LOG_LOOKUP_PROACTIVE = "LOG_LOOKUP_PROACTIVE" MEMORY_RECALL = "MEMORY_RECALL" INFERENCE_ONLY = "INFERENCE_ONLY" CALLBACK_COLLECTOR_PROCESSED = "CALLBACK_COLLECTOR_PROCESSED" CALLBACK_LEDGER_RECONCILED = "CALLBACK_LEDGER_RECONCILED" LOG_LOOKUP_OR_SCHEDULE_HISTORY_VERIFICATION = ( "LOG_LOOKUP_OR_SCHEDULE_HISTORY_VERIFICATION" ) # inbound 수신을 의미하는 강한 표현 (사후조회로 사칭 시 fail) _RECEIVED_PHRASES = [ "callback received", "callback 받았", "callback 도착", "callback 수신", "콜백 도착", "콜백 수신", "콜백 받았", "callback inbound", "inbound callback", "수신 완료", "받음 확인", ] # 회장-facing 세션에서 inbound 가 아니면서 received 표현을 사용하는 것 = 거짓 _INBOUND_SOURCES = frozenset( { SourceAttribution.RECEIVED_INBOUND_THIS_SESSION.value, SourceAttribution.CALLBACK_COLLECTOR_PROCESSED.value, } ) def _normalise(text: str) -> str: return re.sub(r"\s+", " ", text or "").strip().lower() def contains_received_phrase(text: str) -> List[str]: """본문에 inbound 수신 표현이 포함되어 있으면 매칭된 phrase 목록을 돌려준다.""" normalised = _normalise(text) return [p for p in _RECEIVED_PHRASES if p in normalised] def is_inbound_source(source: str) -> bool: """source 가 inbound 수신을 정당화할 수 있는 enum 인지.""" return source in _INBOUND_SOURCES def detect_received_misuse( text: str, source_attribution: Optional[str], ) -> Tuple[bool, List[str]]: """ANCHOR-9 — "received/도착/수신" 표현 + source attribution 없거나 비 inbound 면 violation. Args: text: 종료 직전 마지막 user-facing 출력. source_attribution: 처리 세션이 명시한 source enum (None = 미명시). Returns: (violation, matched_phrases). violation=True 이면 Stop hook 종료 차단. """ matches = contains_received_phrase(text) if not matches: return False, [] if source_attribution is None: return True, matches if not is_inbound_source(source_attribution): return True, matches return False, matches def detect_schedule_history_as_inbound(text: str, source_attribution: Optional[str]) -> bool: """ANCHOR-9 (Stop hook 조건 8) — schedule_history/cron-history 사후 조회를 inbound 수신처럼 표현하면 fail. LOG_LOOKUP_OR_SCHEDULE_HISTORY_VERIFICATION source 인데 received phrase 가 있으면 거짓표현 (1C0F6F52 사건 동일 패턴). """ if source_attribution != ( SourceAttribution.LOG_LOOKUP_OR_SCHEDULE_HISTORY_VERIFICATION.value ): return False return bool(contains_received_phrase(text)) def classify_source( *, inbound_envelope_present: bool, collector_mode: bool, ledger_lookup_only: bool, schedule_history_lookup_only: bool, chair_question_triggered: bool, ) -> str: """현재 세션 컨텍스트로부터 가장 정확한 source enum 을 분류. 우선순위 (회장 verbatim 의도): 1. collector_mode 이고 inbound envelope 있음 → CALLBACK_COLLECTOR_PROCESSED 2. inbound envelope 있음 → RECEIVED_INBOUND_THIS_SESSION 3. ledger 만 조회 → CALLBACK_LEDGER_RECONCILED 4. schedule_history 만 조회 → LOG_LOOKUP_OR_SCHEDULE_HISTORY_VERIFICATION 5. chair 가 물어서 로그 본 경우 → LOG_LOOKUP_AFTER_CHAIR_QUESTION 6. 그 외 자발적 로그 조회 → LOG_LOOKUP_PROACTIVE 7. 그 외 → INFERENCE_ONLY """ if collector_mode and inbound_envelope_present: return SourceAttribution.CALLBACK_COLLECTOR_PROCESSED.value if inbound_envelope_present: return SourceAttribution.RECEIVED_INBOUND_THIS_SESSION.value if ledger_lookup_only: return SourceAttribution.CALLBACK_LEDGER_RECONCILED.value if schedule_history_lookup_only: return SourceAttribution.LOG_LOOKUP_OR_SCHEDULE_HISTORY_VERIFICATION.value if chair_question_triggered: return SourceAttribution.LOG_LOOKUP_AFTER_CHAIR_QUESTION.value return SourceAttribution.INFERENCE_ONLY.value def validate(text: str, source_attribution: Optional[str]) -> Dict[str, object]: """Stop hook 에서 호출하는 정본 진입점. Returns: { "violation": bool, "reason": Optional[str], "matched_phrases": List[str], "source_attribution": Optional[str], } """ misuse, matches = detect_received_misuse(text, source_attribution) if misuse: if source_attribution is None: reason = ( "RECEIVED_PHRASE_WITHOUT_SOURCE_ATTRIBUTION " "(Stop hook 조건 7: '도착/수신' 표현 + source 미명시)" ) else: reason = ( "RECEIVED_PHRASE_NON_INBOUND_SOURCE " f"(source={source_attribution} 은 inbound 아님 — " "사후 조회를 수신으로 흐림)" ) return { "violation": True, "reason": reason, "matched_phrases": matches, "source_attribution": source_attribution, } if detect_schedule_history_as_inbound(text, source_attribution): return { "violation": True, "reason": ( "SCHEDULE_HISTORY_VERIFICATION_AS_INBOUND " "(Stop hook 조건 8: 1C0F6F52 패턴 — chain status=ok 사후 확인을 " "수신처럼 표현)" ), "matched_phrases": contains_received_phrase(text), "source_attribution": source_attribution, } return { "violation": False, "reason": None, "matched_phrases": [], "source_attribution": source_attribution, } __all__ = [ "SCHEMA", "SourceAttribution", "contains_received_phrase", "is_inbound_source", "detect_received_misuse", "detect_schedule_history_as_inbound", "classify_source", "validate", ]