Spaces:

stylevoice
/

talk

Sleeping

talk / backend /services /memory_format.py
user
Deploy StyleVoiceChat backend
aa9892b 4 months ago
56.8 kB
"""사용자 기억 포맷 유틸 — 의미기억(패싯) / 일화기억(날짜 라인) 순수 함수.

I/O 없음(테스트 용이). chat.py 의 병합-저장 경로와 주입 경로, 그리고 Phase 2
컨텍스트 라우터가 공유한다.

설계(req_user_memory_redesign §1·§3 + req_memory_situational_layers §4·§5):
  - 의미기억 = timeless 사실. 패싯 헤더([기본]/[취향·습관]/[이상형]/[성적·패티쉬]/[AI 정보]).
  - 일화기억 = 시점 있는 사건/상태. 신 포맷 "[MM-DD HH:MM] 사건" 한 줄씩(구 "[YYYY-MM-DD]"
    도 graceful 파싱), 순간상태는 끝에 "(상태)".
    * 시점은 서버가 KST 로 스탬프(LLM 은 '지금'을 모름) — 신 포맷은 분 단위까지.
    * 저장: ≤60일 / ≤30줄 (오래된 것부터 버림).
    * 주입: 일반 ≤13일 / 순간상태 ≤1일 만, 절대시점 → 상대시간("방금/3시간 전/어제/지난주") 렌더.
  - 저장 직전 후처리 정규화(§5, LLM 준수율 비의존, 결정적): 빈칸·플레이스홀더 제거 ·
    헤더없는 선두 사실 [기본] 라우팅 · 의미↔일화 누수(순간상태/일회성 사건) 일화로 relocate.
"""
from __future__ import annotations

import re
from datetime import date, datetime, timezone

# --- 일화기억 -----------------------------------------------------------------

EPISODE_MERGE_MARKER = "===일화==="    # 병합 출력 의미기억 ↔ 일화 구분
DIRECTIVE_MERGE_MARKER = "===지침==="  # 병합 출력 일화 ↔ 절차적 지침(proposed) 구분
_MOMENTARY_TAG = "(상태)"            # 순간상태(감기·피곤 등) 태그 — 짧은 디케이
# 신 포맷 `[MM-DD HH:MM @동네]` + 구 포맷 `[YYYY-MM-DD]`/`[YYYY-MM-DD HH:MM]`(장소 없음)
#   모두 파싱(graceful). YYYY 생략(신 포맷)은 기준연도로 추론, 결과가 미래면 연말↔연초 경계에서
#   전년으로 롤오버. `@동네` 는 그 사건 당시 사용자가 있던 동(선택) — 서버가 스탬프(LLM 미기재).
_DATED_RE = re.compile(
    r"^\[(?:(\d{4})-)?(\d{1,2})-(\d{1,2})(?:[ T](\d{1,2}):(\d{2}))?"
    r"(?:\s*@\s*([^\]]+?))?\s*\]\s*(.*)$"
)

EPISODE_STORE_MAX_LINES = 30
EPISODE_STORE_MAX_AGE_DAYS = 60
EPISODE_INJECT_GENERAL_DAYS = 13     # 일반 일화: 이 일수 초과 시 주입 제외(=14일 디케이)
EPISODE_INJECT_MOMENTARY_DAYS = 1    # 순간상태: 이 일수 초과 시 주입 제외(=1~2일 디케이)
EPISODE_HOUR_JOIN = " · "            # 같은 1시간 블록 라인 결합 구분자(세그먼트 dedupe 키)


def _coerce_now(ref: "date | datetime") -> "tuple[datetime, bool]":
    """ref(date|datetime) → (기준 datetime, 시각정보 보유 여부).

    date 만 주면 시각 없음(=날짜 단위). datetime 이면 분 단위까지 사용. 호출측이 KST 의
    '지금'(권장: datetime) 또는 '오늘'(date, 하위호환)을 넘긴다.
    """
    if isinstance(ref, datetime):
        return ref, True
    return datetime(ref.year, ref.month, ref.day), False


def _parse_episode_line(
    line: str, now: "datetime | None" = None
) -> "tuple[datetime | None, bool, str, str]":
    """'[타임스탬프 @동네] 본문' → (datetime, 시각보유, 동네, 본문). 시점 없으면 (None, False, "", 원문).

    - 구 `[YYYY-MM-DD]`/신 `[MM-DD HH:MM]`/장소포함 `[MM-DD HH:MM @고기동]` 모두 허용.
    - `@동네` 는 선택 — 없으면 "". 그 사건 당시 사용자가 있던 동(서버 스탬프).
    - YYYY 생략 시 now 연도로 추론, 추론값이 now 보다 미래면 60일 윈도우 안일 때만 전년 롤오버.
    - now 의 tzinfo 를 결과 datetime 에 부여(naive/aware 혼합 비교 방지). now 미전달 시 파싱만.
    """
    s = line.strip()
    if not s:
        return None, False, "", ""
    m = _DATED_RE.match(s)
    if not m:
        return None, False, "", s
    year, mon, day, hh, mm, place, body = m.groups()
    has_time = hh is not None
    place = (place or "").strip()
    tz = now.tzinfo if now is not None else None
    base_year = int(year) if year is not None else (now.year if now is not None else date.today().year)
    try:
        dt = datetime(
            base_year, int(mon), int(day),
            int(hh) if has_time else 0, int(mm) if has_time else 0,
            tzinfo=tz,
        )
    except ValueError:
        return None, False, "", s
    if now is not None and year is None and dt > now:
        # 연도 추론 결과가 미래 → 최근 윈도우 안이면 전년(연말↔연초 경계), 아니면 그대로
        # 두고 클램프는 호출측(stamp_and_trim)에 맡긴다.
        prev = dt.replace(year=dt.year - 1)
        if (now - prev).days <= EPISODE_STORE_MAX_AGE_DAYS + 1:
            dt = prev
    return dt, has_time, place, (body or "").strip()


def _fmt_stamp(dt: datetime, has_time: bool, place: str = "") -> str:
    """저장 포맷 — 신 `[MM-DD HH:MM @동네]`. 시각 없으면 `[MM-DD @동네]`, 장소 없으면 `@` 생략."""
    place = (place or "").strip()
    p = f" @{place}" if place else ""
    if has_time:
        return f"[{dt.month:02d}-{dt.day:02d} {dt.hour:02d}:{dt.minute:02d}{p}]"
    return f"[{dt.month:02d}-{dt.day:02d}{p}]"


# 성적 대화 일화 차단 (2026-07-14 사용자 지시) — 성적 내용은 의미기억 [성적] 패싯으로만,
#   일화([최근 대화] 히스토리)에는 남기지 않는다. 병합 LLM 프롬프트 지시의 결정적 백스톱.
#   ⚠️ 고정밀 결합형만 — '성적' 단독은 학교 성적·성적표와 충돌, '삽입'은 개발 대화
#   (데이터 삽입), '사정'은 일상어(사정이 생겨서)와 충돌하므로 넣지 않는다.
_SEXUAL_EPISODE_RE = re.compile(
    r"성적[ ]?(묘사|행위|인|으로|취향|긴장|장면|대화|농담|상상|욕구|매력|접촉|판타지)"
    r"|섹스|성행위|성관계|음란|야한|야하게|패티쉬|페티시|자위|신음|애무|전희"
    r"|오르가즘|알몸|나체|성기|음부|유두|가슴을 만지|엉덩이를 만지"
)

_STAMP_PREFIX_RE = re.compile(r"^(\[[^\]]*\])\s*(.*)$")


def strip_sexual_episode_segments(raw: str) -> str:
    """성적 내용 세그먼트/라인 제거 — 스탬프는 남은 세그먼트에 보존.

    한 시간 블록 라인(` · ` 결합)에 일반 화제와 성적 세그먼트가 섞여 있으면
    성적 세그먼트만 걷어내고 나머지는 유지한다. 라인 전체가 성적이면 라인 드랍.
    """
    out: list[str] = []
    for line in (raw or "").splitlines():
        s = line.strip()
        if not s:
            continue
        m = _STAMP_PREFIX_RE.match(s)
        stamp, body = (m.group(1), m.group(2)) if m else ("", s)
        kept = [
            g.strip() for g in body.split(EPISODE_HOUR_JOIN)
            if g.strip() and not _SEXUAL_EPISODE_RE.search(g)
        ]
        if not kept:
            continue
        rebuilt = EPISODE_HOUR_JOIN.join(kept)
        out.append(f"{stamp} {rebuilt}".strip() if stamp else rebuilt)
    return "\n".join(out)


def _norm_seg(s: str) -> str:
    """세그먼트 정규화 — 공백·말미 구두점 무시(중복 판정용 키)."""
    return re.sub(r"\s+", "", s).rstrip(".,!?~")


def _seg_bigrams(s: str) -> set:
    return {s[i:i + 2] for i in range(len(s) - 1)} if len(s) >= 2 else {s}


def _seg_near_dup(a: str, b: str) -> bool:
    """세그먼트 근사-동일 판정 — 보수적(2026-07-14 실데이터 캘리브레이션).

    같은 사건의 재서술(병합 LLM 이 표현만 바꿔 다시 쓴 것)만 흡수한다:
    ① 정규화 후 동일 ② 한쪽이 다른쪽을 포함(같은 내용 + 세부 추가) ③ 문자
    바이그램 Jaccard ≥ 0.75(거의 동일 문장). ⚠️ 임계를 낮추면 반대 의미가 합쳐진다
    — "야근 안 하고 일찍 퇴근" vs "야근 때문에 늦게 퇴근" = 0.458, "신라면에 계란"
    vs "김치찌개에 계란" = 0.609 실측. 0.75 미만의 유사 중복은 발생원 차단(병합
    베이스라인 + 프롬프트 재서술 금지)이 담당한다.
    """
    na, nb = _norm_seg(a), _norm_seg(b)
    if not na or not nb:
        return False
    if na == nb:
        return True
    # 포함 관계 = 같은 진술 + 세부 추가 (예: "라면 먹음" ⊂ "라면 먹음. 계란 넣음").
    #   최소 8자 — 더 짧으면 무관한 긴 문장에 우연히 포함될 위험.
    if len(na) >= 8 and len(nb) >= 8 and (na in nb or nb in na):
        return True
    ba, bb = _seg_bigrams(na), _seg_bigrams(nb)
    union = len(ba | bb)
    return union > 0 and len(ba & bb) / union >= 0.75


def _merge_segs(existing: "list[str]", incoming: "list[str]") -> "list[str]":
    """세그먼트 병합 — 근사-동일은 1개만(더 긴 쪽 유지 = 세부 정보 보존)."""
    out = list(existing)
    for seg in incoming:
        for i, have in enumerate(out):
            if _seg_near_dup(have, seg):
                if len(seg) > len(have):
                    out[i] = seg
                break
        else:
            out.append(seg)
    return out


def _coalesce_hour_blocks(
    parsed: "list[tuple[datetime, bool, str, str]]",
) -> "list[tuple[datetime, bool, str, str]]":
    """같은 (날짜·시(hour)) 라인을 한 줄로 결합 — "1시간 블록 = 한 줄" 저장 규칙.

    병합 LLM 이 발화 단위로 잘게 쪼갠 라인(같은 시각 8줄 등)을 결정적으로 흡수한다
    (LLM 준수율 비의존). 시점·동네는 그 블록 첫(가장 이른) 라인 것을 유지 — 입력이
    시점 오름차순 정렬돼 있다고 가정. 시각 없는(구 [MM-DD]) 라인은 결합하지 않는다
    (하위호환). 본문은 EPISODE_HOUR_JOIN 으로 이어 붙이되 근사-동일 세그먼트는
    1개만(재결합 멱등).

    ⚠️ 2026-07-14 변경: 결합 키에서 동네 제거 — 운전 중엔 동네가 몇 분마다 바뀌어
    "1시간 = 한 줄"이 사실상 작동하지 않았다(같은 드라이브가 2분 간격 여러 줄로
    파편화, 사용자 보고). 동네 궤적은 시간 블록 단위(블록 첫 위치)로 남는다.
    """
    out: list[tuple[datetime, bool, str, str]] = []
    index: dict[tuple[int, int, int, int], int] = {}
    for dt, has_time, loc, body in parsed:
        if not has_time:
            out.append((dt, has_time, loc, body))
            continue
        key = (dt.year, dt.month, dt.day, dt.hour)
        i = index.get(key)
        if i is None:
            index[key] = len(out)
            out.append((dt, has_time, loc, body))
            continue
        pdt, pht, ploc, pbody = out[i]
        segs = [s.strip() for s in pbody.split(EPISODE_HOUR_JOIN) if s.strip()]
        incoming = [s.strip() for s in body.split(EPISODE_HOUR_JOIN) if s.strip()]
        merged = _merge_segs(segs, incoming)
        if merged != segs:
            out[i] = (pdt, pht, ploc, EPISODE_HOUR_JOIN.join(merged))
    return out


def stamp_and_trim_episodes(raw: str, ref: "date | datetime", place: str = "") -> str:
    """새 일화에 KST '지금'(ref)과 현재 장소(place)를 스탬프하고 저장 한도(≤60일·≤30줄)로 정리.

    - 시점이 있는 줄은 그 시점·장소 유지(기존 라인 보존), 신 포맷으로 정규화.
    - '[오늘]' 리터럴 또는 시점 없는 줄(=이번 세션 새 사건)은 ref 로 스탬프하고,
      place 가 있으면 그 동네를 `@동네` 로 함께 박제(그 사건 당시 사용자가 있던 곳).
    - ref 보다 미래 시점은 ref 로 클램프(LLM 환각 방지).
    - 같은 (날짜·시·동네) 라인은 한 줄로 결합(_coalesce_hour_blocks) — "1시간 블록 = 한 줄".
    - 60일 초과 제거, 30줄 초과 시 최신부터 30줄 유지. 같은(시점·본문) 중복 1개만.
    반환은 시점 오름차순(과거→현재) 정규화 텍스트. ref 가 datetime 이면 새 스탬프에 시각 포함.
    place 는 동 토큰만(예: '고기동') 넘기는 걸 권장 — 호출측이 축약해 전달.
    """
    if not (raw or "").strip():
        return ""
    # 성적 대화 일화 차단 (2026-07-14) — 저장 경로 단일 지점에서 결정적으로 필터.
    #   기존 저장분에 남은 성적 라인도 다음 병합 통과 시 함께 정리된다(소급 정화).
    raw = strip_sexual_episode_segments(raw)
    if not raw.strip():
        return ""
    now, now_has_time = _coerce_now(ref)
    new_place = (place or "").strip()
    parsed: list[tuple[datetime, bool, str, str]] = []
    seen: set[tuple[str, str]] = set()
    for line in raw.splitlines():
        s = line.strip()
        if not s:
            continue
        if s.startswith("[오늘]"):
            body = s[len("[오늘]"):].strip()
            dt, has_time, loc = now, now_has_time, new_place
        else:
            dt, has_time, loc, body = _parse_episode_line(s, now)
            if dt is None:
                # 시점 없는 새 줄(relocate 본문 등) → 지금·현재 장소로 스탬프.
                dt, has_time, loc, body = now, now_has_time, new_place, (body or s)
        if dt > now:
            dt, has_time = now, now_has_time  # 미래 스탬프 클램프
        if not body:
            continue
        key = (dt.replace(microsecond=0).isoformat(), body)
        if key in seen:
            continue
        seen.add(key)
        parsed.append((dt, has_time, loc, body))
    if not parsed:
        return ""
    # 60일 초과 제거
    parsed = [t for t in parsed if (now - t[0]).days <= EPISODE_STORE_MAX_AGE_DAYS]
    # 시점 오름차순(동일 시점은 입력 순서 보존 — stable sort)
    parsed.sort(key=lambda t: t[0])
    # 같은 1시간 블록(날짜·시·동네) 결합 — 발화 단위 나열을 한 줄 히스토리로.
    parsed = _coalesce_hour_blocks(parsed)
    # 30줄 초과 시 최신(뒤쪽)부터 30줄 유지
    if len(parsed) > EPISODE_STORE_MAX_LINES:
        parsed = parsed[-EPISODE_STORE_MAX_LINES:]
    return "\n".join(f"{_fmt_stamp(dt, ht, loc)} {b}" for (dt, ht, loc, b) in parsed)


def _relative_label(now: datetime, dt: datetime, has_time: bool) -> str:
    """경과 시점 → 한국어 상대표현.

    같은 날 & 시각정보 있으면 시간 단위("방금/3시간 전"), 아니면 일 단위("오늘/어제/3일 전…").
    일 경계는 24h 가 아니라 달력 일차(day_diff)로 판단 — 자정 넘으면 '어제'.
    """
    day_diff = (now.date() - dt.date()).days
    if day_diff <= 0:
        if has_time:
            hours = (now - dt).total_seconds() / 3600.0
            if hours < 1:
                return "방금"
            return f"{int(hours)}시간 전"
        return "오늘"
    if day_diff == 1:
        return "어제"
    if day_diff <= 6:
        return f"{day_diff}일 전"
    if day_diff <= 13:
        return "지난주"
    if day_diff <= 20:
        return "2주 전"
    if day_diff <= 29:
        return "3주 전"
    if day_diff <= 59:
        return "지난달"
    return "오래전"


def _beat_line(now: datetime, dt: datetime, has_time: bool, place: str, body: str) -> str:
    """주입용 한 줄 — "- 3시간 전 · 대전: 본문". 장소 없으면 " · 장소" 생략."""
    loc = f" · {place}" if (place or "").strip() else ""
    return f"- {_relative_label(now, dt, has_time)}{loc}: {body}"


def render_episodes(raw: str, ref: "date | datetime") -> str:
    """주입용 일화 렌더 — 디케이 필터 + 절대시점 → 상대시간(+ 그때 있던 동네).

    일반 일화는 ≤13일, 순간상태(상태) 태그는 ≤1일 만 통과. 통과한 라인을
    "- 어제 · 고기동: 코스트코 다녀옴" / "- 3시간 전: …" 형태로. 통과 라인 없으면 빈 문자열.
    ref 가 datetime 이고 라인에 시각이 있으면 같은 날은 시간 단위로 렌더.
    """
    if not (raw or "").strip():
        return ""
    now, ref_has_time = _coerce_now(ref)
    out: list[tuple[datetime, bool, str, str]] = []
    for line in raw.splitlines():
        dt, has_time, place, body = _parse_episode_line(line, now)
        if dt is None or not body:
            continue
        day_diff = (now.date() - dt.date()).days
        momentary = _MOMENTARY_TAG in body
        limit = EPISODE_INJECT_MOMENTARY_DAYS if momentary else EPISODE_INJECT_GENERAL_DAYS
        if day_diff > limit:
            continue
        clean = body.replace(_MOMENTARY_TAG, "").strip()
        if not clean:
            continue
        out.append((dt, has_time and ref_has_time, place, clean))
    if not out:
        return ""
    out.sort(key=lambda t: t[0])  # 과거 → 현재
    return "\n".join(_beat_line(now, dt, ht, loc, b) for (dt, ht, loc, b) in out)


# --- 의미기억 패싯 라우팅 (Phase 2 컨텍스트 라우터) -------------------------
# 패싯 헤더별 게이트. 알 수 없는 헤더(옛 [사용자 정보]/[최근 우리 얘기] 등)는 항상 주입
# (graceful — 마이그레이션 전 블롭은 통째 주입되다 다음 병합 때 패싯으로 재배치된다).

FACET_ALWAYS = ("기본", "취향·습관", "AI 정보")    # 항상 주입(+ [최근대화] 별도 경로)
FACET_BASE = FACET_ALWAYS                            # 하위호환 별칭

# --- 캐릭터별 관계 기억 키잉 (req_character_relation_memory §P2) --------------
# 활성 캐릭터 카드 id 기준으로 AI 관계 섹션을 분리: 키 ''(기본/무카드) = 기존
#   [AI 정보] 그대로(하위호환 100%), 키 X = [AI:X]. 주입·병합 모두 "활성 것 하나만".
#   ⚠️ "알 수 없는 헤더 항상 주입" 규칙에 [AI: prefix 만 예외 — 그 외 unknown 동작 불변.
AI_BASE_HEADER = "AI 정보"
AI_KEYED_PREFIX = "AI:"


def ai_header_for_key(active_ai_key: str) -> str:
    """활성 캐릭터 키 → AI 관계 섹션 헤더(대괄호 제외). '' = 기존 [AI 정보]."""
    k = (active_ai_key or "").strip()
    return f"{AI_KEYED_PREFIX}{k}" if k else AI_BASE_HEADER


def is_ai_header(header: str) -> bool:
    """AI 관계 섹션 헤더인가 — 기본 [AI 정보] 또는 캐릭터 키 [AI:*]."""
    h = (header or "").strip()
    return h == AI_BASE_HEADER or h.startswith(AI_KEYED_PREFIX)


def sanitize_active_ai_key(key, characters) -> str:
    """persona 의 active_character_id 정합성 — characters 목록에 없는 id 는 ''(기본).

    구 클라이언트/구 데이터(키 부재)는 '' 취급(기존 [AI 정보] 경로, 회귀 0).
    """
    k = str(key or "").strip()
    if not k:
        return ""
    for c in characters or []:
        if isinstance(c, dict) and str(c.get("id") or "").strip() == k:
            return k
    return ""
# 상황별(게이팅) 버킷 — 활성 집합에 있을 때만 주입(req §2 택소노미).
FACET_TIME = ("아침", "밤")                          # 시각 휴리스틱
FACET_ACTIVITY = ("회사·일", "퇴근후", "식사")        # 시각+요일 휴리스틱(운전은 별도 라우트)
FACET_DRIVING = ("운전",)                            # 운전모드(stream_tts) 라우트
FACET_AFFECTION = ("이상형", "애정")                  # 연애맥락(spice≥중간)만. 애정 추가(P2)
# spice==hot 만. 신 헤더 [성적](req §2) + 옛 [성적·패티쉬] 둘 다 인식 → 미재병합 데이터도
# 게이트·lazy 주입 정상(회귀 0). 신규 출력/정규화는 [성적] 으로 수렴.
FACET_HOT = ("성적", "성적·패티쉬")
# 모든 게이팅 버킷 — 활성 집합에 없으면 주입 제외(+ 스코프 병합 시 held-out 대상).
_ALL_GATED = (
    set(FACET_TIME) | set(FACET_ACTIVITY) | set(FACET_DRIVING)
    | set(FACET_AFFECTION) | set(FACET_HOT)
)
_GATED_FACETS = _ALL_GATED   # 하위호환 별칭

# 헤더 별칭 — 옛 표기를 신 정규 헤더로 그래이스풀 흡수(normalize 에서 적용). 본문은 보존.
_FACET_HEADER_ALIASES = {"성적·패티쉬": "성적", "성적·패티시": "성적"}

# [성적] 6 하위필드(req §2). 빈 칸은 P0 normalize 의 플레이스홀더 제거가 정리.
SEXUAL_SUBFIELDS = (
    "대상·취향", "자세·행위", "상황·장면", "대화톤·진행", "호칭·어휘", "수위·경계",
)

# --- 관계 이정표 (req_character_relation_memory §P3-2) -------------------------
# AI 관계 섹션 끝의 `이정표:` 하위 리스트 — 관계 전환점(첫 만남/호칭 변화/약속 등)의
#   append-only 로그. 병합 LLM 은 기존 줄 수정·삭제 금지, 새 줄은 `[오늘]` → 서버가
#   [MM-DD] 날짜 스탬프(일화 스탬프 패턴 재사용). 상한 10줄(초과 backstop = 오래된 것부터).
MILESTONE_LABEL = "이정표:"
MILESTONE_MAX_LINES = 10


def _split_milestone_block(body: str) -> "tuple[list[str], list[str]]":
    """AI 섹션 본문 → (이정표 이전 라인들, 이정표 라인들). 라벨 없으면 (전체, [])."""
    lines = body.splitlines()
    for i, ln in enumerate(lines):
        if _strip_bullet(ln.strip()) == MILESTONE_LABEL:
            head = [l for l in lines[:i]]
            tail = [l for l in lines[i + 1:] if l.strip()]
            return head, tail
    return lines, []


def milestone_lines(body: str) -> "list[str]":
    """AI 섹션 본문의 이정표 라인(strip 정규화). 회고 검증·검사용."""
    return [ln.strip() for ln in _split_milestone_block(body)[1]]


def stamp_milestones(semantic: str, ref: "date | datetime") -> str:
    """AI 관계 섹션 이정표의 새 줄(`[오늘]`/무표기)에 [MM-DD] 스탬프 + 상한 10줄 backstop.

    기존 스탬프가 있는 줄은 **원문 그대로**(append-only — 표기 재작성 금지). 상한 초과는
    오래된(위쪽) 줄부터 드롭 — "덜 중요한 줄 제거" 큐레이션은 병합 LLM 몫, 이건 결정적 안전망.
    """
    now, _ = _coerce_now(ref)
    order: list[str] = []
    out: dict[str, str] = {}
    for header, body in split_facets(semantic).items():
        if is_ai_header(header):
            head, tail = _split_milestone_block(body)
            if tail:
                stamped: list[str] = []
                for ln in tail:
                    core = _strip_bullet(ln.strip())
                    if core.startswith("[오늘]"):
                        rest = core[len("[오늘]"):].strip()
                        if rest:
                            stamped.append(f"- {_fmt_stamp(now, False)} {rest}")
                        continue
                    dt, _ht, _loc, _b = _parse_episode_line(core, now)
                    if dt is not None:
                        stamped.append(ln)   # 기존 스탬프 줄 = 원문 보존
                    else:
                        stamped.append(f"- {_fmt_stamp(now, False)} {core}")
                if len(stamped) > MILESTONE_MAX_LINES:
                    stamped = stamped[-MILESTONE_MAX_LINES:]
                parts = [l for l in head if l.strip()]
                if stamped:
                    parts += [MILESTONE_LABEL, *stamped]
                body = "\n".join(parts).strip()
        if body.strip():
            out[header] = body
            order.append(header)
    return "\n".join(
        (f"[{h}]\n{out[h]}" if h else out[h]) for h in order
    ).strip()

_FACET_HEADER_RE = re.compile(r"^\[([^\]]+)\]\s*$")


def _canonical_facet_header(h: str) -> str:
    """헤더 표기 정규화 — 별칭([성적·패티쉬]→[성적]) 흡수."""
    h = (h or "").strip()
    return _FACET_HEADER_ALIASES.get(h, h)


def split_facets(semantic: str) -> "dict[str, str]":
    """패싯 블롭 → {헤더(대괄호 제외): 본문}. 입력 순서 보존. 헤더 없는 선두 라인은 키 ''."""
    sections: dict[str, list[str]] = {}
    order: list[str] = []
    cur = ""
    for line in (semantic or "").splitlines():
        m = _FACET_HEADER_RE.match(line.strip())
        if m:
            cur = m.group(1).strip()
            if cur not in sections:
                sections[cur] = []
                order.append(cur)
            continue
        if cur not in sections:
            sections[cur] = []
            order.append(cur)
        sections[cur].append(line)
    return {k: "\n".join(sections[k]).strip() for k in order}


def render_facets(semantic: str, *, affection: bool, hot: bool) -> str:
    """게이트에 맞는 패싯만 골라 재조립.

    base([기본]/[취향·습관]/[AI 정보]) + 알 수 없는 헤더는 항상. [이상형] 은 affection,
    [성적·패티쉬] 는 hot 일 때만. 빈 본문 섹션은 생략.
    """
    out: list[str] = []
    for header, body in split_facets(semantic).items():
        if not body.strip():
            continue
        if header in FACET_HOT and not hot:
            continue
        if header in FACET_AFFECTION and not affection:
            continue
        out.append(f"[{header}]\n{body}" if header else body)
    return "\n".join(out).strip()


def render_facets_subset(semantic: str, headers) -> str:
    """주어진 헤더 집합의 패싯만 재조립 (spice 상승 시 lazy 주입용). 빈 섹션 생략."""
    want = set(headers)
    out: list[str] = []
    for header, body in split_facets(semantic).items():
        if header in want and body.strip():
            out.append(f"[{header}]\n{body}")
    return "\n".join(out).strip()


def render_facets_gated(semantic: str, active_buckets, active_ai_key: str = "") -> str:
    """FACET_ALWAYS + 활성 상황 버킷만 주입(req §3 라우터). 비활성 게이트 버킷은 제외.

    미정의(옛 [사용자 정보] 등) 헤더는 항상 주입(graceful — 다음 병합 때 재배치). 빈 섹션 생략.
    active_buckets = 이번 상황의 활성 버킷 헤더 집합(시각/활동/운전/spice 로 산출).
    active_ai_key = 활성 캐릭터 카드 id — AI 관계 섹션([AI 정보]·[AI:*])은 활성 것
    하나만 주입(캐릭터별 관계 기억, req_character_relation_memory §P2). '' = [AI 정보].
    """
    want = set(active_buckets or ())
    active_ai = ai_header_for_key(active_ai_key)
    out: list[str] = []
    for header, body in split_facets(semantic).items():
        if not body.strip():
            continue
        if is_ai_header(header) and header != active_ai:
            continue   # 비활성 캐릭터 관계 기억 유출 차단 ([AI: prefix 만 unknown 예외)
        if header in _ALL_GATED and header not in want:
            continue
        out.append(f"[{header}]\n{body}" if header else body)
    return "\n".join(out).strip()


def active_time_activity_buckets(now: datetime) -> "list[str]":
    """시각·요일 휴리스틱으로 활성 시간/활동 버킷(req §3). GPS 학습(P3) 전 휴리스틱만.

    - 시간대: [아침] 05–10시 · [밤] 22–05시 (오후/저녁은 버킷 없음).
    - 활동(평일): [회사·일] 09–18시 · [퇴근후] 17:30–19시. 그 외(집)=버킷 없음.
    - [식사]: 점심 11:30–13:30 / 저녁 18–20시(요일 무관, 노이즈 가능 — 실사용 후 on/off §10).
    운전 버킷은 라우트(stream_tts)로만 — 여기 포함 안 함.
    """
    out: list[str] = []
    h, mins = now.hour, now.hour * 60 + now.minute
    weekday = now.weekday() < 5
    if 5 <= h < 10:
        out.append("아침")
    if h >= 22 or h < 5:
        out.append("밤")
    if weekday:
        if 9 * 60 <= mins < 18 * 60:
            out.append("회사·일")
        if 17 * 60 + 30 <= mins < 19 * 60:
            out.append("퇴근후")
    if (11 * 60 + 30 <= mins < 13 * 60 + 30) or (18 * 60 <= mins < 20 * 60):
        out.append("식사")
    return out


RECENT_BEATS_CAP = 20


def render_recent_beats(raw: str, ref: "date | datetime", *, cap: int = RECENT_BEATS_CAP) -> str:
    """[최근대화] — 디케이 대신 최근 cap 개 비트 항상 주입(req §4).

    render_episodes 와 달리 디케이 필터 없음 — 시점 최신순 cap 개를 골라 과거→현재로 렌더.
    절대시점 → 상대시간("방금/3시간 전/어제/지난주") + 그때 있던 동네(" · 대전"). (상태)
    태그는 표시에서 제거(현재 단정 말라는 가이드는 주입 wrapper 가 담당). 통과 라인 없으면 빈 문자열.
    """
    if not (raw or "").strip():
        return ""
    now, ref_has_time = _coerce_now(ref)
    parsed: list[tuple[datetime, bool, str, str]] = []
    for line in raw.splitlines():
        dt, has_time, place, body = _parse_episode_line(line, now)
        if dt is None or not body:
            continue
        clean = body.replace(_MOMENTARY_TAG, "").strip()
        if not clean:
            continue
        parsed.append((dt, has_time and ref_has_time, place, clean))
    if not parsed:
        return ""
    parsed.sort(key=lambda t: t[0])   # 과거 → 현재
    parsed = parsed[-cap:]            # 최근 cap 개
    return "\n".join(_beat_line(now, dt, ht, loc, b) for (dt, ht, loc, b) in parsed)


# --- GPS 동네 학습 (req_memory_situational_layers §3·P3) ----------------------
# 시각·요일 휴리스틱을 GPS 학습으로 보정/override. "집 동네/회사 동네"를 빈도로 추론해,
#   평일이라도 집 동네면 [집](활동 버킷 없음), 주말이라도 회사 동네면 [회사·일].
#   GPS 없음/학습 부족/opt-out 이면 active_time_activity_buckets 휴리스틱 그대로(graceful).

LEARN_MIN_OBS = 3   # 집/회사로 인정할 최소 관측 수(콜드스타트 오인 방지)


def classify_learn_bucket(now: datetime) -> "str | None":
    """관측 시각 → 학습 시간대('home'/'work'/None). geo/report 누적용.

    평일 09–18 = work, 평일 저녁·새벽(19–07) + 주말 = home, 평일 통근(07–09/18–19) = 모호 skip.
    앱 사용이 집중되는 저녁·주말(집)과 평일 코어(회사)에서만 신호를 모은다.
    """
    h = now.hour
    weekday = now.weekday() < 5
    if weekday and 9 <= h < 18:
        return "work"
    if (weekday and (h >= 19 or h < 7)) or (not weekday):
        return "home"
    return None


def _place_token(s: str) -> str:
    """동 단위 비교 키 — 역지오코딩 문자열의 가장 구체적(마지막) 토큰(예: '고기동')."""
    s = (s or "").strip()
    return s.split()[-1] if s else ""


def _same_place(a: str, b: str) -> bool:
    """두 위치가 같은 동인가 — 전체 일치 또는 동 토큰 일치(접두 미세변형 흡수)."""
    a, b = (a or "").strip(), (b or "").strip()
    if not a or not b:
        return False
    if a == b:
        return True
    ta, tb = _place_token(a), _place_token(b)
    return bool(ta) and ta == tb


def infer_home_work(learned: "dict | None") -> "tuple[str | None, str | None]":
    """학습 데이터 → (집 동네, 회사 동네) 전체 문자열. 최소 관측 미달이면 None.

    집 = home 최다, 회사 = work 최다(집과 다른 동). 동률·미달은 None(휴리스틱 폴백).
    """
    dongs = ((learned or {}).get("dongs") or {})
    if not isinstance(dongs, dict) or not dongs:
        return None, None

    def _best(field: str, exclude: "str | None") -> "str | None":
        best, best_n = None, 0
        for place, counts in dongs.items():
            if exclude is not None and _same_place(place, exclude):
                continue
            n = int((counts or {}).get(field, 0) or 0)
            if n >= LEARN_MIN_OBS and n > best_n:
                best, best_n = place, n
        return best

    home = _best("home", None)
    work = _best("work", home)   # 회사는 집과 다른 동
    return home, work


def resolve_situation_buckets(
    now: datetime, current_place: str, learned: "dict | None"
) -> "list[str]":
    """현재 GPS 동 + 학습으로 P2 시각휴리스틱을 보정한 활성 상황 버킷.

    - 현재 동이 회사 동네 → [회사·일](+퇴근시간대면 [퇴근후]). 시간/식사 버킷은 시계 그대로.
    - 현재 동이 집 동네 → 활동 버킷 없음(집). 시간(아침/밤)·식사만.
    - 매칭 없음/GPS 없음/학습 부족 → active_time_activity_buckets 휴리스틱 그대로(graceful).
    """
    base = active_time_activity_buckets(now)
    current_place = (current_place or "").strip()
    if not current_place:
        return base
    home, work = infer_home_work(learned)
    if work is None and home is None:
        return base
    # 시간/식사 버킷(시계 확정) 은 override 후에도 유지, 활동(회사·일/퇴근후) 만 GPS 가 정정.
    time_meal = [b for b in base if b in ("아침", "밤", "식사")]
    if work and _same_place(current_place, work):
        act = ["회사·일"]
        mins = now.hour * 60 + now.minute
        if 17 * 60 + 30 <= mins < 19 * 60:
            act.append("퇴근후")
        return time_meal + act
    if home and _same_place(current_place, home):
        return time_meal   # 집 = 활동 버킷 없음
    return base


def record_observation(learned: "dict | None", place: str, now: datetime,
                       *, max_dongs: int = 12) -> "dict | None":
    """학습 객체에 관측 1건 누적(순수함수) → 갱신본 반환. 변화 없으면 None.

    place(동 문자열)·now(KST) 의 시간대(classify_learn_bucket)에 카운트 +1.
    top max_dongs 동만 보관(총빈도 낮은 동 prune). 모호 시간대/빈 place 는 무시(None).
    카운트는 안전 상한(1e6)으로 클램프. updated_at 갱신.
    """
    place = (place or "").strip()
    bucket = classify_learn_bucket(now)
    if not place or bucket is None:
        return None
    out = dict(learned or {})
    dongs = dict(out.get("dongs") or {})
    # 동 토큰으로 기존 키 병합(접두 미세변형 흡수) — 같은 동이면 기존 키 재사용.
    key = place
    for existing in dongs:
        if _same_place(existing, place):
            key = existing
            break
    counts = dict(dongs.get(key) or {})
    counts[bucket] = min(int(counts.get(bucket, 0) or 0) + 1, 1_000_000)
    counts.setdefault("home", int(counts.get("home", 0) or 0))
    counts.setdefault("work", int(counts.get("work", 0) or 0))
    dongs[key] = counts
    # top N prune — 총빈도(home+work) 낮은 동 제거.
    if len(dongs) > max_dongs:
        ranked = sorted(
            dongs.items(),
            key=lambda kv: (kv[1].get("home", 0) + kv[1].get("work", 0)),
            reverse=True,
        )
        dongs = dict(ranked[:max_dongs])
    out["dongs"] = dongs
    out["updated_at"] = now.astimezone(timezone.utc).isoformat()
    return out


def parse_merge_sections(text: str) -> tuple[str, str, str]:
    """병합 LLM 출력 → (의미기억, 일화 raw, 지침 raw).

    기대 순서: 의미기억 → ===일화=== → 일화 → ===지침=== → 지침. 마커 부재는 graceful
    (해당 섹션 빈 문자열). 의미기억은 항상 가장 앞 마커 이전 = 지침 파싱 실패가 의미기억을
    오염시키지 못한다(회귀 격리).
    """
    t = (text or "").strip()
    if not t:
        return "", "", ""
    ep = t.find(EPISODE_MERGE_MARKER)
    dr = t.find(DIRECTIVE_MERGE_MARKER)
    present = [i for i in (ep, dr) if i >= 0]
    sem_end = min(present) if present else len(t)
    semantic = t[:sem_end].strip()
    episodes = ""
    directives = ""
    if ep >= 0:
        ep_start = ep + len(EPISODE_MERGE_MARKER)
        ep_end = dr if (dr >= 0 and dr > ep) else len(t)
        episodes = t[ep_start:ep_end].strip()
    if dr >= 0:
        directives = t[dr + len(DIRECTIVE_MERGE_MARKER):].strip()
    return semantic, episodes, directives


def parse_merge_output(text: str) -> tuple[str, str]:
    """(의미기억, 일화 raw) — parse_merge_sections 의 앞 2개(하위호환)."""
    semantic, episodes, _ = parse_merge_sections(text)
    return semantic, episodes


# --- 저장 직전 후처리 정규화 (req_memory_situational_layers §5) ----------------
# LLM 준수율에 비의존하는 결정적 가드. **삭제가 아니라 정리/이동**(additive·graceful):
#   회귀 가드(§7) — 의미기억 사실을 임의 삭제하지 않는다. 누수는 일화로 relocate(디케이로
#   소멸하되 영속이면 다음 병합이 재학습). 사용자↔AI 분리는 병합 프롬프트 가드와 병행.

# 빈칸/플레이스홀더 — 정보 없는 줄(저장 노이즈)·"(없음)" 류 제거용.
_PLACEHOLDER_VALUES = {
    "", "없음", "비공개", "미상", "모름", "해당없음", "해당 없음", "정보없음", "정보 없음",
    "n/a", "na", "tbd", "?", "??", "...", "…", "-", "—", "x",
}
# [기본]의 정규 라벨 — 영속 사실이므로 누수 relocate 대상에서 보호(순간상태 키워드가 섞여도
#   통째로 옮기지 않는다). 라벨 표기 변형 흡수용.
_BASE_CANONICAL_LABELS = {
    "이름", "호칭", "호칭/말투", "말투", "신체", "가족", "사람", "가족/사람",
    "직업", "직업/일", "사는 곳", "사는곳", "거주지", "사는 곳/직장",
}
# 순간상태 키워드 — 의미기억에 누수된 일시적 신체/컨디션(일화로 이동, (상태) 태그 부여).
#   짧은 토큰은 오탐(예: "아프"⊂"아프리카")을 피해 구체형으로.
_MOMENTARY_HINTS = (
    "뻐근", "결리", "쑤시", "아파", "아프다", "아픈데", "아픔", "아팠", "따끔",
    "부었", "붓기", "감기 기운", "감기 걸", "몸살", "열나", "미열", "피곤", "졸리",
    "졸려", "배고프", "배불", "취했", "숙취", "더부룩", "메스꺼", "어지러",
    "컨디션 난조", "몸이 안",
)
# 일회성 사건 키워드 — 의미기억에 누수된 과거 사건(일화로 이동). 습관(좋아함/탐/씀/다님 등)은
#   비해당 — 과거시제 행위 어미만 매칭해 영속 취향 오인 최소화.
_EVENT_HINTS = (
    "먹었", "마셨", "다녀왔", "다녀옴", "갔다왔", "갔다 왔", "구경했",
    "들렀", "구매했", "주문했", "보러 갔", "만나고 왔", "보고 왔",
)
_LEAK_SCANNED_FACETS = ("기본", "취향·습관")
_BULLET_CHARS = "-•·*∙▪ \t"


def _strip_bullet(s: str) -> str:
    return s.lstrip(_BULLET_CHARS).strip()


def _is_placeholder_value(val: str) -> bool:
    v = val.strip()
    # 템플릿 힌트 echo 방지 — 값 전체가 괄호로 감싸인 안내문(예: "(연애 상대 선호)",
    #   "(무엇/누구에게 끌리는지)", "(한 줄씩)")은 실제 내용이 아니다. 실사실은 단일 괄호로
    #   통째 감싸 쓰지 않는다("누나 (주로)" 처럼 부분 괄호는 실값 → 보존).
    if len(v) >= 2 and v[0] == "(" and v[-1] == ")":
        return True
    return v.strip("()[]").strip().lower() in _PLACEHOLDER_VALUES


def _clean_facet_body(body: str) -> str:
    """빈/플레이스홀더 불릿 제거. 'label: value' 는 value 가, 단독 불릿은 본문이 기준.

    원래 들여쓰기/불릿 표기는 보존(첫 strip 후 그대로 유지)해 포맷 변형 최소화.
    '이정표:' 라벨 줄은 값이 없어도 보존(하위 리스트의 컨테이너) — 단, 아래에 이정표가
    한 줄도 없으면 마지막에 정리(빈 헤더 저장 차단과 동일 원칙).
    """
    kept: list[str] = []
    for line in body.splitlines():
        s = line.strip()
        if not s:
            continue
        core = _strip_bullet(s)
        if not core:
            continue
        if core == MILESTONE_LABEL:
            kept.append(s)   # 이정표 컨테이너 라벨 — 플레이스홀더 판정 제외
            continue
        if ":" in core:
            _, _, val = core.partition(":")
            if _is_placeholder_value(val):
                continue
        elif _is_placeholder_value(core):
            continue
        kept.append(s)
    # 하위 이정표가 전부 정리돼 라벨만 남으면 라벨도 제거.
    if kept and _strip_bullet(kept[-1]) == MILESTONE_LABEL:
        kept.pop()
    return "\n".join(kept).strip()


def normalize_semantic(semantic: str) -> str:
    """의미기억 결정적 정리 — 헤더 별칭·헤더없는 선두 [기본] 라우팅 + 빈/플레이스홀더 제거.

    - 헤더 별칭 정규화([성적·패티쉬]→[성적]). 별칭 충돌 시 본문 병합(삭제 금지).
    - split_facets 의 헤더없는 선두('' 키)는 [기본] 으로 흡수(없으면 신설, 있으면 앞에 병합).
    - 빈 본문이 된 패싯(헤더만 남는)은 생략 → "[이상형] (없음)" 같은 플레이스홀더 저장 차단.
    - 미정의(옛 [사용자 정보] 등) 헤더는 보존 — graceful 재배치는 병합 LLM 의 몫(회귀 가드).
    """
    facets = split_facets(semantic)
    lead = facets.get("", "").strip()
    order: list[str] = []
    cleaned: dict[str, str] = {}
    for header, body in facets.items():
        if header == "":
            continue
        canon = _canonical_facet_header(header)
        cb = _clean_facet_body(body)
        if not cb:
            continue
        if canon in cleaned:
            cleaned[canon] = cleaned[canon] + "\n" + cb   # 별칭 충돌 병합
        else:
            cleaned[canon] = cb
            order.append(canon)
    if lead:
        lead_clean = _clean_facet_body(lead)
        if lead_clean:
            if "기본" in cleaned:
                cleaned["기본"] = lead_clean + "\n" + cleaned["기본"]
            else:
                cleaned["기본"] = lead_clean
                order.insert(0, "기본")
    return "\n".join(f"[{h}]\n{cleaned[h]}" for h in order).strip()


def _relocate_leaked_episodes(semantic: str) -> "tuple[str, list[str]]":
    """[기본]/[취향·습관] 에 누수된 순간상태·일회성 사건을 분리.

    반환: (정리된 의미기억, 일화로 옮길 본문 리스트). **삭제가 아니라 이동** — 호출측이
    리스트를 일화로 스탬프한다. [기본] 의 정규 라벨(이름·신체·가족 등 영속)은 보호.
    순간상태는 끝에 (상태) 태그를 붙여 짧은 디케이가 걸리게 한다.
    """
    facets = split_facets(semantic)
    leaked: list[str] = []
    order: list[str] = []
    out: dict[str, str] = {}
    for header, body in facets.items():
        if header not in _LEAK_SCANNED_FACETS:
            out[header] = body
            order.append(header)
            continue
        kept: list[str] = []
        for line in body.splitlines():
            s = line.strip()
            if not s:
                continue
            core = _strip_bullet(s)
            label, value = "", core
            if ":" in core:
                label, _, value = core.partition(":")
                label, value = label.strip(), value.strip()
            # [기본] 정규 라벨은 영속 사실 → 보호(누수 판정 제외).
            if header == "기본" and label in _BASE_CANONICAL_LABELS:
                kept.append(s)
                continue
            momentary = any(k in core for k in _MOMENTARY_HINTS)
            event = any(k in core for k in _EVENT_HINTS)
            if momentary or event:
                txt = value or core
                if momentary and _MOMENTARY_TAG not in txt:
                    txt = f"{txt} {_MOMENTARY_TAG}"
                leaked.append(txt)
            else:
                kept.append(s)
        kb = "\n".join(kept).strip()
        if kb:
            out[header] = kb
            order.append(header)
    sem2 = "\n".join(
        (f"[{h}]\n{out[h]}" if h else out[h]) for h in order if out.get(h, "").strip()
    ).strip()
    return sem2, leaked


def normalize_merge_result(semantic: str, episodes_raw: str) -> "tuple[str, str]":
    """저장 직전 후처리 종합 — 의미기억 정리 + 의미↔일화 누수 relocate. 순수함수.

    반환된 episodes_raw 의 추가 라인은 시점이 없어 호출측 stamp_and_trim_episodes 가
    '지금'으로 스탬프(미래 클램프·≤60일 트림도 거기서 일괄). 사용자↔AI 분리는 병합 프롬프트
    가드와 병행 — 여기서는 결정적으로 안전한 항목만 정리(영속 사실 임의 삭제 금지, 회귀 §7).
    """
    sem = normalize_semantic(semantic)
    sem, leaked = _relocate_leaked_episodes(sem)
    ep = (episodes_raw or "").strip()
    if leaked:
        extra = "\n".join(leaked)
        ep = (ep + "\n" + extra).strip() if ep else extra
    return sem, ep


# --- 자동 캡 — 버킷 유형별 소프트 예산 (req_memory_situational_layers §5) ---------
# 전역 상한 제거 → 버킷별 소프트 예산. 초과 시 약한/오래된(=뒤쪽) 줄부터 whole-line 트림.
#   저장 총량은 키우되(상황별 게이팅으로 1회 주입량은 별도 제어) 무한 성장은 막는다(§1).
#   ⚠️ 캡은 길이 가드일 뿐 — 절대 줄을 쪼개 자르지 않는다(불릿 단위). 안전(하드리밋) 라인 보호.
FACET_SOFT_BUDGET = {
    "기본": 400,
    "AI 정보": 600,          # 400→600 (관계 단계 + 이정표 포함, req_character_relation §P3-3)
    "취향·습관": 700,
    "이상형": 500,
    "성적": 1000,
    "성적·패티쉬": 1000,   # 별칭 안전망(정규화 전 잔존 데이터)
}
FACET_DEFAULT_BUDGET = 700   # 미정의(옛/상황) 헤더 — 너그럽게
# 트림에서 보호할 라벨 — 성적 하드리밋/금지선은 안전상 마지막까지 유지.
_CAP_PROTECTED_LABELS = {"수위·경계", "수위", "경계", "하드리밋", "하드 리밋", "금지선", "절대 금지"}


def facet_budget(header: str) -> int:
    """패싯 소프트 예산 — AI 관계 섹션([AI 정보]·[AI:*])은 공통 600, 그 외 표 참조."""
    if is_ai_header(header):
        return FACET_SOFT_BUDGET[AI_BASE_HEADER]
    return FACET_SOFT_BUDGET.get(header, FACET_DEFAULT_BUDGET)


def _line_label(line: str) -> str:
    core = _strip_bullet(line)
    return core.split(":", 1)[0].strip() if ":" in core else ""


def _cap_facet_body(body: str, budget: int, *, protect_milestones: bool = False) -> str:
    """본문이 budget(자) 초과면 비보호 라인을 뒤(약한/최신)에서부터 whole-line 드롭.

    보호 라벨(수위·경계 등 안전선)은 마지막까지 유지. 모든 라인이 보호 라인이면 그대로 둔다
    (안전 우위 — 길이보다 하드리밋 보존 우선). protect_milestones(AI 관계 섹션)이면
    `이정표:` 블록은 통상 트림에서 제외 — 비보호 라인이 소진됐을 때만 오래된 이정표부터
    마지막 수단으로 드롭(append-only 우선 보존).
    """
    if len(body) <= budget:
        return body
    lines = [ln for ln in body.splitlines() if ln.strip()]
    ms: set[int] = set()
    if protect_milestones:
        for i, ln in enumerate(lines):
            if _strip_bullet(ln.strip()) == MILESTONE_LABEL:
                ms = set(range(i, len(lines)))   # 라벨 + 이정표 전부 보호
                break
    while len("\n".join(lines)) > budget:
        drop_idx = None
        for i in range(len(lines) - 1, -1, -1):
            if i in ms:
                continue
            if _line_label(lines[i]) not in _CAP_PROTECTED_LABELS:
                drop_idx = i
                break
        if drop_idx is None:
            ms_sorted = sorted(ms)
            if len(ms_sorted) >= 2:
                drop_idx = ms_sorted[1]   # [0]=라벨 — 가장 오래된 이정표부터
            else:
                break  # 보호 라인만 남음 → 더 못 줄임
        del lines[drop_idx]
        ms = {i - 1 if i > drop_idx else i for i in ms if i != drop_idx}
    # 이정표가 전부 드롭돼 라벨만 남으면 라벨도 정리.
    if lines and _strip_bullet(lines[-1].strip()) == MILESTONE_LABEL:
        lines.pop()
    return "\n".join(lines).strip()


def cap_facets(semantic: str) -> str:
    """패싯별 소프트 예산 적용(전역 상한 대체). 빈 패싯 생략. 헤더 순서 보존.

    예산은 facet_budget(AI 관계 섹션 공통 600, 미정의 헤더 FACET_DEFAULT_BUDGET). 순수함수.
    """
    order: list[str] = []
    out: dict[str, str] = {}
    for header, body in split_facets(semantic).items():
        capped = _cap_facet_body(
            body, facet_budget(header), protect_milestones=is_ai_header(header)
        )
        if capped.strip():
            out[header] = capped
            order.append(header)
    return "\n".join((f"[{h}]\n{out[h]}" if h else out[h]) for h in order).strip()


# --- 스코프 병합 (req_memory_situational_layers §5) ----------------------------
# 활성 버킷만 재병합 — 입력·출력을 활성 버킷으로 한정해 16K자 전체 재emit(비용·드리프트)
#   회피. 비활성 게이트 버킷(예: 비-성적 세션의 [성적])은 held-out → LLM 에 안 보여
#   재작성 불가, 병합 후 verbatim 재부착(append-only — 새 사실만 보태고 기존 줄은 절대 손실 X).

def _norm_line(line: str) -> str:
    return _strip_bullet(line).strip()


def split_active_inactive(semantic: str, active_buckets) -> "tuple[str, dict[str, str]]":
    """(editable 의미기억, held_out 패싯 dict).

    editable = FACET_ALWAYS + 활성 게이트 버킷 + 미정의(옛) 헤더(graceful 재배치 대상).
    held_out = 게이트 버킷 중 비활성 = LLM 에 안 보냄(드리프트 차단), 병합 후 재부착.
    """
    want = set(FACET_ALWAYS) | set(active_buckets or ())
    order: list[str] = []
    editable: dict[str, str] = {}
    held: dict[str, str] = {}
    for header, body in split_facets(semantic).items():
        if header in _ALL_GATED and header not in want:
            held[header] = body
        else:
            editable[header] = body
            order.append(header)
    editable_text = "\n".join(
        (f"[{h}]\n{editable[h]}" if h else editable[h]) for h in order if editable.get(h, "").strip()
    ).strip()
    return editable_text, held


def recombine_scoped(merged_semantic: str, held_out: "dict[str, str]") -> str:
    """병합된 활성 의미기억 + held_out(비활성, verbatim) 재결합. **append-only**.

    held_out 라인은 절대 잃지 않는다. merged 가 held_out 헤더에 새 라인을 emit 한 경우
    (LLM 이 비활성 버킷 사실을 추론) 중복 아닌 줄만 보탠다. 헤더 별칭 정규화 적용.
    """
    order: list[str] = []
    out: dict[str, str] = {}

    def _add(header: str, body: str) -> None:
        h = _canonical_facet_header(header)
        body = (body or "").strip()
        if not body:
            return
        if h in out:
            seen = {_norm_line(ln) for ln in out[h].splitlines() if ln.strip()}
            extra = [ln for ln in body.splitlines() if ln.strip() and _norm_line(ln) not in seen]
            if extra:
                out[h] = (out[h] + "\n" + "\n".join(extra)).strip()
        else:
            out[h] = body
            order.append(h)

    for header, body in split_facets(merged_semantic).items():
        _add(header, body)
    for header, body in (held_out or {}).items():
        _add(header, body)
    return "\n".join((f"[{h}]\n{out[h]}" if h else out[h]) for h in order).strip()


# --- 캐릭터별 AI 관계 섹션 분리/정리 (req_character_relation_memory §P2) --------

def split_ai_sections(semantic: str, active_ai_key: str = "") -> "tuple[str, dict[str, str]]":
    """(활성 AI 섹션만 남긴 의미기억, held_out 비활성 AI 섹션 dict).

    split_active_inactive 와 동일 패턴 — 비활성 AI 관계 섹션([AI 정보] 포함, 활성 헤더
    제외 전부)은 병합 LLM 에 안 보여주고(드리프트 차단) 병합 후 verbatim 재부착한다.
    """
    active_h = ai_header_for_key(active_ai_key)
    order: list[str] = []
    editable: dict[str, str] = {}
    held: dict[str, str] = {}
    for header, body in split_facets(semantic).items():
        if is_ai_header(header) and header != active_h:
            held[header] = body
        else:
            editable[header] = body
            order.append(header)
    text = "\n".join(
        (f"[{h}]\n{editable[h]}" if h else editable[h])
        for h in order if editable.get(h, "").strip()
    ).strip()
    return text, held


def remove_facet_sections(semantic: str, headers) -> str:
    """주어진 헤더 섹션을 블롭에서 제거 — 카드 삭제 시 [AI:id] 관계 기억 삭제용."""
    drop = {(h or "").strip() for h in (headers or ())}
    order: list[str] = []
    out: dict[str, str] = {}
    for header, body in split_facets(semantic).items():
        if header in drop:
            continue
        if body.strip():
            out[header] = body
            order.append(header)
    return "\n".join((f"[{h}]\n{out[h]}" if h else out[h]) for h in order).strip()


def strip_keyed_ai_sections(semantic: str) -> str:
    """[AI:*] 캐릭터 관계 섹션만 제거 — 라디오 DJ 등 비대화 경로 교차 오염 차단.

    기존 [AI 정보] 는 유지(현행 동작 불변). 캐릭터별 관계 기억은 자유대화·운전
    persona 경로 전용(⛔ _dj_base_persona 교차 오염 금지, req §5).
    """
    order: list[str] = []
    out: dict[str, str] = {}
    for header, body in split_facets(semantic).items():
        if header.startswith(AI_KEYED_PREFIX):
            continue
        if body.strip():
            out[header] = body
            order.append(header)
    return "\n".join((f"[{h}]\n{out[h]}" if h else out[h]) for h in order).strip()


# --- 압력 기반 회고 (req_character_relation_memory §P3-4) ----------------------
# 트리거 = "이번 병합으로 길이가 실제 증가 && 캡의 80% 초과"인 패싯만(입력 압력 기반,
#   시간 크론 없음). 증가 없으면 스킵 — 압축 불가능한 패싯의 매병합 재발동 방지.
#   회고는 언제나 "없어도 되는 최적화" — 검증 실패 = 결과 폐기·원본 유지(기억 손실 금지).

REFLECT_TRIGGER_RATIO = 0.8


def facets_needing_reflection(old_semantic: str, new_semantic: str) -> "list[str]":
    """회고 대상 패싯 헤더 — 병합 전보다 길이가 실제 증가했고 캡 80% 를 넘은 것만."""
    old = split_facets(old_semantic or "")
    out: list[str] = []
    for header, body in split_facets(new_semantic or "").items():
        if not header or not body.strip():
            continue
        old_len = len((old.get(header) or "").strip())
        new_len = len(body.strip())
        if new_len > old_len and new_len > facet_budget(header) * REFLECT_TRIGGER_RATIO:
            out.append(header)
    return out


def facets_over_budget(semantic: str, ratio: float = REFLECT_TRIGGER_RATIO) -> "list[str]":
    """캡 대비 ratio 초과 패싯(성장 조건 없음) — 시간 기반 회고 크론용.

    2026-07-10 사용자 확정으로 원 설계(§2 '압력 기반만')에 매일 1회 크론을 보완:
    병합으로는 더 안 자라(압력 트리거가 안 걸리는) 과밀 패싯도 하루 1회 회고 대상.
    검증·폐기 안전장치(validate_reflection)는 동일하게 적용된다.
    """
    out: list[str] = []
    for header, body in split_facets(semantic or "").items():
        b = (body or "").strip()
        if header and b and len(b) > facet_budget(header) * ratio:
            out.append(header)
    return out


def validate_reflection(header: str, old_body: str, reflected_text: str) -> "tuple[bool, str]":
    """회고 LLM 출력 검증 → (통과 여부, 정제된 새 본문).

    코드 레벨 검사(§P3-4): ① 헤더 보존(그 헤더 하나만) ② 기존 이정표 줄 전부 원문 포함
    ③ 압축만 허용(본문 길이 증가 금지). 하나라도 실패 = (False, "") — 호출측이 원본 유지.
    """
    t = (reflected_text or "").strip()
    if not t:
        return False, ""
    facets = split_facets(t)
    named = [h for h in facets if h]
    if named != [header]:
        return False, ""   # 헤더 소실/변경/다른 헤더 발명
    if (facets.get("") or "").strip():
        return False, ""   # 헤더 밖 선두 본문 = 형식 위반
    new_body = (facets.get(header) or "").strip()
    if not new_body:
        return False, ""
    old_body = (old_body or "").strip()
    if len(new_body) > len(old_body):
        return False, ""   # 압축이 아니라 증식 — 새 사실 생성 의심
    old_ms = milestone_lines(old_body)
    if old_ms:
        new_lines = {ln.strip() for ln in new_body.splitlines() if ln.strip()}
        if any(m not in new_lines for m in old_ms):
            return False, ""   # 이정표 원문 누락 = 폐기
    return True, new_body


def replace_facet_body(semantic: str, header: str, new_body: str) -> str:
    """한 패싯의 본문만 교체(헤더 순서 보존) — 회고 결과 반영용."""
    order: list[str] = []
    out: dict[str, str] = {}
    for h, body in split_facets(semantic).items():
        b = new_body if h == header else body
        if b.strip():
            out[h] = b.strip()
            order.append(h)
    return "\n".join((f"[{h}]\n{out[h]}" if h else out[h]) for h in order).strip()

"""사용자 기억 포맷 유틸 — 의미기억(패싯) / 일화기억(날짜 라인) 순수 함수.
I/O 없음(테스트 용이). chat.py 의 병합-저장 경로와 주입 경로, 그리고 Phase 2
컨텍스트 라우터가 공유한다.
설계(req_user_memory_redesign §1·§3 + req_memory_situational_layers §4·§5):
  - 의미기억 = timeless 사실. 패싯 헤더([기본]/[취향·습관]/[이상형]/[성적·패티쉬]/[AI 정보]).
  - 일화기억 = 시점 있는 사건/상태. 신 포맷 "[MM-DD HH:MM] 사건" 한 줄씩(구 "[YYYY-MM-DD]"
    도 graceful 파싱), 순간상태는 끝에 "(상태)".
    * 시점은 서버가 KST 로 스탬프(LLM 은 '지금'을 모름) — 신 포맷은 분 단위까지.
    * 저장: ≤60일 / ≤30줄 (오래된 것부터 버림).
    * 주입: 일반 ≤13일 / 순간상태 ≤1일 만, 절대시점 → 상대시간("방금/3시간 전/어제/지난주") 렌더.
  - 저장 직전 후처리 정규화(§5, LLM 준수율 비의존, 결정적): 빈칸·플레이스홀더 제거 ·
    헤더없는 선두 사실 [기본] 라우팅 · 의미↔일화 누수(순간상태/일회성 사건) 일화로 relocate.
"""
from __future__ import annotations
import re
from datetime import date, datetime, timezone
# --- 일화기억 -----------------------------------------------------------------
EPISODE_MERGE_MARKER = "===일화==="    # 병합 출력 의미기억 ↔ 일화 구분
DIRECTIVE_MERGE_MARKER = "===지침==="  # 병합 출력 일화 ↔ 절차적 지침(proposed) 구분
_MOMENTARY_TAG = "(상태)"            # 순간상태(감기·피곤 등) 태그 — 짧은 디케이
# 신 포맷 `[MM-DD HH:MM @동네]` + 구 포맷 `[YYYY-MM-DD]`/`[YYYY-MM-DD HH:MM]`(장소 없음)
#   모두 파싱(graceful). YYYY 생략(신 포맷)은 기준연도로 추론, 결과가 미래면 연말↔연초 경계에서
#   전년으로 롤오버. `@동네` 는 그 사건 당시 사용자가 있던 동(선택) — 서버가 스탬프(LLM 미기재).
_DATED_RE = re.compile(
    r"^\[(?:(\d{4})-)?(\d{1,2})-(\d{1,2})(?:[ T](\d{1,2}):(\d{2}))?"
    r"(?:\s*@\s*([^\]]+?))?\s*\]\s*(.*)$"
)
EPISODE_STORE_MAX_LINES = 30
EPISODE_STORE_MAX_AGE_DAYS = 60
EPISODE_INJECT_GENERAL_DAYS = 13     # 일반 일화: 이 일수 초과 시 주입 제외(=14일 디케이)
EPISODE_INJECT_MOMENTARY_DAYS = 1    # 순간상태: 이 일수 초과 시 주입 제외(=1~2일 디케이)
EPISODE_HOUR_JOIN = " · "            # 같은 1시간 블록 라인 결합 구분자(세그먼트 dedupe 키)
def _coerce_now(ref: "date | datetime") -> "tuple[datetime, bool]":
    """ref(date|datetime) → (기준 datetime, 시각정보 보유 여부).
    date 만 주면 시각 없음(=날짜 단위). datetime 이면 분 단위까지 사용. 호출측이 KST 의
    '지금'(권장: datetime) 또는 '오늘'(date, 하위호환)을 넘긴다.
    """
    if isinstance(ref, datetime):
        return ref, True
    return datetime(ref.year, ref.month, ref.day), False
def _parse_episode_line(
    line: str, now: "datetime | None" = None
) -> "tuple[datetime | None, bool, str, str]":
    """'[타임스탬프 @동네] 본문' → (datetime, 시각보유, 동네, 본문). 시점 없으면 (None, False, "", 원문).
    - 구 `[YYYY-MM-DD]`/신 `[MM-DD HH:MM]`/장소포함 `[MM-DD HH:MM @고기동]` 모두 허용.
    - `@동네` 는 선택 — 없으면 "". 그 사건 당시 사용자가 있던 동(서버 스탬프).
    - YYYY 생략 시 now 연도로 추론, 추론값이 now 보다 미래면 60일 윈도우 안일 때만 전년 롤오버.
    - now 의 tzinfo 를 결과 datetime 에 부여(naive/aware 혼합 비교 방지). now 미전달 시 파싱만.
    """
    s = line.strip()
    if not s:
        return None, False, "", ""
    m = _DATED_RE.match(s)
    if not m:
        return None, False, "", s
    year, mon, day, hh, mm, place, body = m.groups()
    has_time = hh is not None
    place = (place or "").strip()
    tz = now.tzinfo if now is not None else None
    base_year = int(year) if year is not None else (now.year if now is not None else date.today().year)
    try:
        dt = datetime(
            base_year, int(mon), int(day),
            int(hh) if has_time else 0, int(mm) if has_time else 0,
            tzinfo=tz,
        )
    except ValueError:
        return None, False, "", s
    if now is not None and year is None and dt > now:
        # 연도 추론 결과가 미래 → 최근 윈도우 안이면 전년(연말↔연초 경계), 아니면 그대로
        # 두고 클램프는 호출측(stamp_and_trim)에 맡긴다.
        prev = dt.replace(year=dt.year - 1)
        if (now - prev).days <= EPISODE_STORE_MAX_AGE_DAYS + 1:
            dt = prev
    return dt, has_time, place, (body or "").strip()
def _fmt_stamp(dt: datetime, has_time: bool, place: str = "") -> str:
    """저장 포맷 — 신 `[MM-DD HH:MM @동네]`. 시각 없으면 `[MM-DD @동네]`, 장소 없으면 `@` 생략."""
    place = (place or "").strip()
    p = f" @{place}" if place else ""
    if has_time:
        return f"[{dt.month:02d}-{dt.day:02d} {dt.hour:02d}:{dt.minute:02d}{p}]"
    return f"[{dt.month:02d}-{dt.day:02d}{p}]"
# 성적 대화 일화 차단 (2026-07-14 사용자 지시) — 성적 내용은 의미기억 [성적] 패싯으로만,
#   일화([최근 대화] 히스토리)에는 남기지 않는다. 병합 LLM 프롬프트 지시의 결정적 백스톱.