From b401085518bbb00d8d04fc6874faca82fa7c530c Mon Sep 17 00:00:00 2001
From: Hyungi Ahn <hyungiahn@Hyungiui-MacBookPro.local>
Date: Fri, 24 Apr 2026 09:30:42 +0900
Subject: [PATCH] =?UTF-8?q?feat(ai):=20EscalationEnvelope=20contract=20(4B?=
 =?UTF-8?q?=E2=86=9226B=20handoff)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

frozen dataclass with from_stage / escalation_reasons / risk_flags /
distilled_context / original_pointers / synthesis_directives / user_intent /
draft_hint. JSON round-trip (to_json/from_json). to_system_injection() 으로
26B system prompt 에 주입할 텍스트 블록 생성 (risk_flags + directives +
distilled_context 순).

from_stage 는 whitelist 검증 (triage/classify/summarize_short/advice_trigger/
night_sweep/ask_pre/unknown). tuple 타입 강제 (mutability 방지).

PR-B 의 escalation_service 가 이 계약을 사용. PR-A 는 계약만 정의.

plan: ~/.claude/plans/wise-gliding-hippo.md

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 app/ai/envelope.py | 97 ++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 97 insertions(+)
 create mode 100644 app/ai/envelope.py

diff --git a/app/ai/envelope.py b/app/ai/envelope.py
new file mode 100644
index 0000000..c0c6330
--- /dev/null
+++ b/app/ai/envelope.py
@@ -0,0 +1,97 @@
+"""EscalationEnvelope — 4B → 26B 핸드오프 계약.
+
+4B 가 "자신이 처리 못한다" 고 판단했을 때 26B 에게 전달하는 구조화 메시지.
+26B 는 distilled_context 로 방향을 잡고 original_pointers 로 필요한 원문만 재조회.
+
+PR-A 는 dataclass 계약만 정의. 실제 생성/소비는 PR-B 의 escalation_service 가 담당.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import asdict, dataclass, field
+from typing import Any
+
+
+ValidFromStage = {
+    "triage",
+    "classify",
+    "summarize_short",
+    "advice_trigger",
+    "night_sweep",
+    "ask_pre",
+    "unknown",  # 호환성용
+}
+
+
+@dataclass(frozen=True)
+class EscalationEnvelope:
+    from_stage: str
+    escalation_reasons: tuple[str, ...]
+    risk_flags: tuple[str, ...]
+    distilled_context: str
+    original_pointers: dict[str, Any] = field(default_factory=dict)
+    synthesis_directives: tuple[str, ...] = ()
+    user_intent: str | None = None
+    draft_hint: str | None = None
+
+    def __post_init__(self) -> None:
+        if self.from_stage not in ValidFromStage:
+            raise ValueError(
+                f"from_stage '{self.from_stage}' not in {ValidFromStage}"
+            )
+        if not isinstance(self.escalation_reasons, tuple):
+            raise TypeError("escalation_reasons must be tuple (for hashability)")
+        if not isinstance(self.risk_flags, tuple):
+            raise TypeError("risk_flags must be tuple (for hashability)")
+        if not isinstance(self.synthesis_directives, tuple):
+            raise TypeError("synthesis_directives must be tuple (for hashability)")
+
+    # -- 26B system prompt 주입용 텍스트 -----------------------------------
+    def to_system_injection(self) -> str:
+        lines = [
+            "=== ESCALATION ENVELOPE (from 4B) ===",
+            f"from_stage: {self.from_stage}",
+            f"reasons: {', '.join(self.escalation_reasons) or '(none)'}",
+            f"risk_flags: {', '.join(self.risk_flags) or '(none)'}",
+        ]
+        if self.user_intent:
+            lines.append(f"user_intent: {self.user_intent}")
+        if self.draft_hint:
+            lines.append(f"draft_hint: {self.draft_hint}")
+
+        if self.synthesis_directives:
+            lines.append("")
+            lines.append("synthesis_directives (각 risk_flag 별 지시사항, 반드시 준수):")
+            for d in self.synthesis_directives:
+                lines.append(f"  - {d}")
+
+        if self.distilled_context:
+            lines.append("")
+            lines.append("distilled_context (4B 가 압축한 요지 — 참고용, 숫자·인용은 원문 재확인 필수):")
+            lines.append(self.distilled_context)
+
+        if self.original_pointers:
+            lines.append("")
+            lines.append("original_pointers (필요 시 재조회):")
+            lines.append(json.dumps(self.original_pointers, ensure_ascii=False, indent=2))
+
+        return "\n".join(lines)
+
+    # -- JSON round-trip ---------------------------------------------------
+    def to_json(self) -> str:
+        return json.dumps(asdict(self), ensure_ascii=False)
+
+    @classmethod
+    def from_json(cls, s: str) -> EscalationEnvelope:
+        raw = json.loads(s)
+        return cls(
+            from_stage=raw["from_stage"],
+            escalation_reasons=tuple(raw.get("escalation_reasons", ())),
+            risk_flags=tuple(raw.get("risk_flags", ())),
+            distilled_context=raw.get("distilled_context", ""),
+            original_pointers=raw.get("original_pointers", {}) or {},
+            synthesis_directives=tuple(raw.get("synthesis_directives", ())),
+            user_intent=raw.get("user_intent"),
+            draft_hint=raw.get("draft_hint"),
+        )