/agent/memory/summarizer.py

"""
Memory flush manager with Deep Dream distillation

Handles memory persistence when conversation context is trimmed or overflows:
- Uses LLM to summarize discarded messages into concise daily records
- Writes to daily memory files (lazy creation)
- Deduplicates trim flushes to avoid repeated writes
- Runs summarization asynchronously to avoid blocking normal replies
- Deep Dream: periodically distills daily memories → refined MEMORY.md + dream diary
"""

import threading
from typing import Optional, Callable, Any, List, Dict
from pathlib import Path
from datetime import datetime
from common.log import logger


SUMMARIZE_SYSTEM_PROMPT_ZH = """你是一个对话记录助手。请将对话内容归纳为当天的日常记录。

## 要求

按「事件」维度归纳发生的事，不要按对话轮次逐条记录：
- 每条一行，用 "- " 开头
- 合并同一件事的多轮对话
- 只记录有意义的事件，忽略闲聊和问候
- 保留关键的决策、结论和待办事项

当对话没有任何记录价值（仅含问候或无意义内容），直接回复"无"。"""

SUMMARIZE_SYSTEM_PROMPT_EN = """You are a conversation-logging assistant. Summarize the conversation into a daily record.

## Requirements

Summarize by "event", not turn by turn:
- One item per line, starting with "- "
- Merge multiple turns about the same thing
- Only record meaningful events; ignore small talk and greetings
- Keep key decisions, conclusions and to-dos

If the conversation has no record value (only greetings or meaningless content), reply with exactly "None"."""

SUMMARIZE_USER_PROMPT_ZH = """请归纳以下对话的日常记录：

{conversation}"""

SUMMARIZE_USER_PROMPT_EN = """Summarize the daily record of the following conversation:

{conversation}"""

# ---------------------------------------------------------------------------
# Deep Dream prompts — distill daily memories → MEMORY.md + dream diary
# ---------------------------------------------------------------------------

DREAM_SYSTEM_PROMPT_ZH = """你是一个记忆整理助手，负责定期整理用户的长期记忆。

你将收到两份材料：
1. **当前长期记忆** — MEMORY.md 的全部现有内容
2. **今日日记** — 当天的日常记录

MEMORY.md 会注入每次对话的系统提示词中，因此必须保持精炼，只存放有价值和值得记忆的内容。

**重要：只能基于提供的材料进行整理，严禁编造、推测或添加材料中不存在的信息。**

## 任务

### Part 1: 更新后的长期记忆（[MEMORY]）

在现有记忆基础上进行整理和提炼，输出完整的更新后内容：
- **合并提炼**：将含义相近的多条合并为一条高密度表述，而非简单罗列
- **新增萃取**：从今日日记中提取值得永久记住的新信息（偏好、决策、人物、规则、经验）
- **冲突更新**：当新信息与旧条目矛盾时，以新信息为准，替换旧条目
- **清理无效**：删除临时性记录、空白条目、格式残留、无意义、重复内容等
- **删除冗余**：已被更精炼表述涵盖的旧条目应删除，避免信息重复
- 每条一行，用 "- " 开头，不带日期前缀
- 可用 "## 标题" 对相关条目分组，使结构更清晰
- 目标：控制在 50 条以内，每条尽量一句话概括

### Part 2: 梦境日记（[DREAM]）

用简洁的叙事风格写一篇短日记，记录这次整理的发现，保持格式美观易读：
- 发现了哪些重复或矛盾
- 从日记中提取了什么新洞察
- 做了哪些清理和优化
- 整体感受和观察

## 输出格式（严格遵守）

```
[MEMORY]
- 记忆条目1
- 记忆条目2
...

[DREAM]
梦境日记内容...
```"""

DREAM_SYSTEM_PROMPT_EN = """You are a memory-curation assistant that periodically organizes the user's long-term memory.

You will receive two inputs:
1. **Current long-term memory** — the full existing content of MEMORY.md
2. **Today's diary** — the daily records

MEMORY.md is injected into the system prompt of every conversation, so it must stay concise and hold only valuable, memory-worthy content.

**Important: organize strictly based on the provided material. Never fabricate, infer, or add information not present in it.**

## Tasks

### Part 1: Updated long-term memory ([MEMORY])

Organize and distill on top of the existing memory, and output the complete updated content:
- **Merge & distill**: combine semantically similar items into one dense statement rather than listing them
- **Extract new**: pull memory-worthy new info from today's diary (preferences, decisions, people, rules, lessons)
- **Resolve conflicts**: when new info contradicts an old item, prefer the new and replace the old
- **Clean invalid**: remove temporary notes, blank items, formatting residue, meaningless or duplicate content
- **Drop redundancy**: delete old items already covered by a more concise statement
- One item per line, starting with "- ", without a date prefix
- You may group related items under "## headings" for clarity
- Goal: keep under 50 items, each ideally a single sentence

### Part 2: Dream diary ([DREAM])

Write a short diary in a concise narrative style recording what this curation found, keep it clean and readable:
- Which duplicates or conflicts were found
- What new insights were extracted from the diary
- What cleanup and optimization was done
- Overall feelings and observations

## Output format (follow strictly)

```
[MEMORY]
- memory item 1
- memory item 2
...

[DREAM]
dream diary content...
```"""

DREAM_USER_PROMPT_ZH = """## 当前长期记忆（MEMORY.md）

{memory_content}

## 近期日记（最近 {days} 天）

{daily_content}"""

DREAM_USER_PROMPT_EN = """## Current long-term memory (MEMORY.md)

{memory_content}

## Recent diary (last {days} days)

{daily_content}"""


def _is_en() -> bool:
    """True when the resolved UI language is English."""
    try:
        from common import i18n
        return i18n.get_language() == "en"
    except Exception:
        return False


def _summarize_system_prompt() -> str:
    return SUMMARIZE_SYSTEM_PROMPT_EN if _is_en() else SUMMARIZE_SYSTEM_PROMPT_ZH


def _summarize_user_prompt() -> str:
    return SUMMARIZE_USER_PROMPT_EN if _is_en() else SUMMARIZE_USER_PROMPT_ZH


def _dream_system_prompt() -> str:
    return DREAM_SYSTEM_PROMPT_EN if _is_en() else DREAM_SYSTEM_PROMPT_ZH


def _dream_user_prompt() -> str:
    return DREAM_USER_PROMPT_EN if _is_en() else DREAM_USER_PROMPT_ZH


def _is_empty_sentinel(text: str) -> bool:
    """Match the "no record value" sentinel in both zh ("无") and en ("None")."""
    if not text:
        return True
    s = text.strip()
    return s == "" or s == "无" or s.lower() == "none"



class MemoryFlushManager:
    """
    Manages memory flush operations.
    
    Flush is triggered by agent_stream in two scenarios:
    1. Context trim: _trim_messages discards old turns → flush discarded content
    2. Context overflow: API rejects request → emergency flush before clearing
    
    Additionally, create_daily_summary() can be called by scheduler for end-of-day summaries.
    """
    
    def __init__(
        self,
        workspace_dir: Path,
        llm_model: Optional[Any] = None,
    ):
        self.workspace_dir = workspace_dir
        self.llm_model = llm_model
        
        self.memory_dir = workspace_dir / "memory"
        self.memory_dir.mkdir(parents=True, exist_ok=True)
        
        self.last_flush_timestamp: Optional[datetime] = None
        self._trim_flushed_hashes: set = set()  # Content hashes of already-flushed messages
        self._last_flushed_content_hash: str = ""  # Content hash at last flush, for daily dedup
        self._last_dream_input_hash: str = ""  # "{date}:{daily_hash}" of last dream, for dedup
        self._last_flush_thread: Optional[threading.Thread] = None
    
    def get_today_memory_file(self, user_id: Optional[str] = None, ensure_exists: bool = False) -> Path:
        """Get today's memory file path: memory/YYYY-MM-DD.md"""
        today = datetime.now().strftime("%Y-%m-%d")
        
        if user_id:
            user_dir = self.memory_dir / "users" / user_id
            if ensure_exists:
                user_dir.mkdir(parents=True, exist_ok=True)
            today_file = user_dir / f"{today}.md"
        else:
            today_file = self.memory_dir / f"{today}.md"
        
        if ensure_exists and not today_file.exists():
            today_file.parent.mkdir(parents=True, exist_ok=True)
            today_file.write_text(f"# Daily Memory: {today}\n\n")
        
        return today_file
    
    def get_main_memory_file(self, user_id: Optional[str] = None) -> Path:
        """Get main memory file path: MEMORY.md (workspace root)"""
        if user_id:
            user_dir = self.memory_dir / "users" / user_id
            user_dir.mkdir(parents=True, exist_ok=True)
            return user_dir / "MEMORY.md"
        else:
            return Path(self.workspace_dir) / "MEMORY.md"
    
    def get_status(self) -> dict:
        return {
            'last_flush_time': self.last_flush_timestamp.isoformat() if self.last_flush_timestamp else None,
            'today_file': str(self.get_today_memory_file()),
            'main_file': str(self.get_main_memory_file())
        }

    # ---- Flush execution (called by agent_stream or scheduler) ----
    
    def flush_from_messages(
        self,
        messages: List[Dict],
        user_id: Optional[str] = None,
        reason: str = "trim",
        max_messages: int = 0,
        context_summary_callback: Optional[Callable[[str], None]] = None,
    ) -> bool:
        """
        Asynchronously summarize and flush messages to daily memory.

        Deduplication runs synchronously, then LLM summarization + file write
        run in a background thread so the main reply flow is never blocked.

        If *context_summary_callback* is provided, it is called with the
        [DAILY] portion of the LLM summary once available. The caller can use
        this to inject the summary into the live message list for context
        continuity — one LLM call serves both disk persistence and in-context
        injection.
        """
        try:
            # Strip scheduler-injected pairs before any further processing.
            # These messages already serve as short-term context inside the
            # receiver session; promoting them into long-term daily memory
            # produces low-value flat logs (e.g. "11:28 price=1013, normal /
            # 11:58 price=1013, normal / ...") and wastes summarisation tokens.
            messages = self._strip_scheduler_pairs(messages)
            if not messages:
                return False

            import hashlib
            deduped = []
            for m in messages:
                text = self._extract_text_from_content(m.get("content", ""))
                if not text or not text.strip():
                    continue
                h = hashlib.md5(text.encode("utf-8")).hexdigest()
                if h not in self._trim_flushed_hashes:
                    self._trim_flushed_hashes.add(h)
                    deduped.append(m)
            if not deduped:
                return False

            import copy
            snapshot = copy.deepcopy(deduped)
            thread = threading.Thread(
                target=self._flush_worker,
                args=(snapshot, user_id, reason, max_messages, context_summary_callback),
                daemon=True,
            )
            thread.start()
            logger.info(f"[MemoryFlush] Async flush dispatched (reason={reason}, msgs={len(snapshot)})")
            self._last_flush_thread = thread
            return True

        except Exception as e:
            logger.warning(f"[MemoryFlush] Failed to dispatch flush (reason={reason}): {e}")
            return False

    def _flush_worker(
        self,
        messages: List[Dict],
        user_id: Optional[str],
        reason: str,
        max_messages: int,
        context_summary_callback: Optional[Callable[[str], None]] = None,
    ):
        """Background worker: summarize with LLM, write daily memory file."""
        try:
            raw_summary = self._summarize_messages(messages, max_messages)
            if _is_empty_sentinel(raw_summary):
                logger.info(f"[MemoryFlush] No valuable content to flush (reason={reason})")
                return

            # Strip legacy [DAILY]/[MEMORY] markers if model still outputs them
            daily_part = self._clean_summary_output(raw_summary)
            if not daily_part:
                return

            # --- Write daily memory ---
            daily_file = ensure_daily_memory_file(self.workspace_dir, user_id)

            headers = {
                "overflow": f"## Context Overflow Recovery ({datetime.now().strftime('%H:%M')})",
                "trim": f"## Trimmed Context ({datetime.now().strftime('%H:%M')})",
                "daily_summary": f"## Daily Summary ({datetime.now().strftime('%H:%M')})",
            }
            header = headers.get(reason, f"## Session Notes ({datetime.now().strftime('%H:%M')})")

            with open(daily_file, "a", encoding="utf-8") as f:
                f.write(f"\n{header}\n\n{daily_part}\n")

            logger.info(f"[MemoryFlush] Wrote daily memory to {daily_file.name} (reason={reason}, chars={len(daily_part)})")

            # --- Inject context summary into live messages (if callback provided) ---
            if context_summary_callback:
                try:
                    context_summary_callback(daily_part)
                except Exception as e:
                    logger.warning(f"[MemoryFlush] Context summary callback failed: {e}")

            self.last_flush_timestamp = datetime.now()

        except Exception as e:
            logger.warning(f"[MemoryFlush] Async flush failed (reason={reason}): {e}")

    @staticmethod
    def _clean_summary_output(raw: str) -> str:
        """Strip legacy [DAILY]/[MEMORY] markers if present, return clean daily text."""
        raw = raw.strip()
        if _is_empty_sentinel(raw):
            return ""

        # Strip [DAILY] marker
        if "[DAILY]" in raw:
            start = raw.index("[DAILY]") + len("[DAILY]")
            end = raw.index("[MEMORY]") if "[MEMORY]" in raw else len(raw)
            raw = raw[start:end].strip()

        # Remove stray [MEMORY] section entirely
        if "[MEMORY]" in raw:
            raw = raw[:raw.index("[MEMORY]")].strip()

        # Remove markdown code fences
        raw = raw.replace("```", "").strip()

        return raw

    def create_daily_summary(
        self,
        messages: List[Dict],
        user_id: Optional[str] = None
    ) -> bool:
        """
        Generate end-of-day summary. Called by daily timer.
        Skips if messages haven't changed since last flush.
        """
        import hashlib
        content = "".join(
            self._extract_text_from_content(m.get("content", ""))
            for m in messages
        )
        content_hash = hashlib.md5(content.encode("utf-8")).hexdigest()
        if content_hash == self._last_flushed_content_hash:
            logger.debug("[MemoryFlush] Daily summary skipped: no new content since last flush")
            return False
        self._last_flushed_content_hash = content_hash
        return self.flush_from_messages(
            messages=messages,
            user_id=user_id,
            reason="daily_summary",
            max_messages=0,
        )

    # ---- Deep Dream (memory distillation) ----

    def deep_dream(self, user_id: Optional[str] = None, lookback_days: int = 1, force: bool = False) -> bool:
        """
        Distill recent daily memories into MEMORY.md and generate a dream diary.

        Args:
            lookback_days: How many days of daily files to read (default 1 for scheduled, 3 for manual)
            force: Skip input-hash dedup check (used by manual /memory dream trigger)
        """
        # Config guard for scheduled runs. Manual trigger (force=True) always
        # runs since it is an explicit user action.
        if not force:
            try:
                from config import conf
                if not conf().get("deep_dream_enabled", True):
                    logger.info("[DeepDream] deep_dream_enabled=false, skipping")
                    return False
            except Exception:
                pass

        if not self.llm_model:
            logger.warning("[DeepDream] No LLM model available, skipping")
            return False

        logger.info(f"[DeepDream] Starting memory distillation (lookback={lookback_days} days)")

        # Collect materials
        memory_content = self._read_main_memory(user_id)
        daily_content, has_content = self._read_recent_dailies(user_id, lookback_days)

        if not has_content:
            logger.info("[DeepDream] No recent daily records, skipping to preserve existing MEMORY.md")
            return False

        # Dedup: skip if same daily content already dreamed today.
        # Note: only hash daily_content (not memory_content), because deep_dream
        # itself rewrites MEMORY.md as a side effect, which would otherwise
        # invalidate the hash on every subsequent call within the same window.
        import hashlib
        daily_hash = hashlib.md5(daily_content.encode("utf-8")).hexdigest()
        today_str = datetime.now().strftime("%Y-%m-%d")
        dedup_key = f"{today_str}:{daily_hash}"
        if not force and dedup_key == self._last_dream_input_hash:
            logger.info("[DeepDream] Already dreamed today with same daily content, skipping")
            return False
        self._last_dream_input_hash = dedup_key

        logger.info(
            f"[DeepDream] Materials collected: "
            f"MEMORY.md={len(memory_content)} chars, "
            f"daily={len(daily_content)} chars"
        )

        # Call LLM for distillation
        import time as _time
        t0 = _time.monotonic()
        try:
            user_msg = _dream_user_prompt().format(
                memory_content=memory_content or "(empty)",
                days=lookback_days,
                daily_content=daily_content or "(no recent daily records)",
            )
            from agent.protocol.models import LLMRequest
            # No output cap: the prompt already keeps MEMORY.md concise (~50
            # items), so a hard max_tokens would only risk truncating a large
            # rewrite. Let the model use its default output budget.
            request = LLMRequest(
                messages=[{"role": "user", "content": user_msg}],
                temperature=0.3,
                stream=False,
                system=_dream_system_prompt(),
            )
            response = self.llm_model.call(request)
            raw = self._extract_response_text(response)
            elapsed = _time.monotonic() - t0
            if not raw or not raw.strip():
                logger.warning(f"[DeepDream] LLM returned empty response ({elapsed:.1f}s)")
                return False
            logger.info(f"[DeepDream] LLM distillation completed ({elapsed:.1f}s, {len(raw)} chars)")
        except Exception as e:
            elapsed = _time.monotonic() - t0
            logger.warning(f"[DeepDream] LLM call failed ({elapsed:.1f}s): {e}")
            return False

        # Parse [MEMORY] and [DREAM] sections
        new_memory, dream_diary = self._parse_dream_output(raw)

        if not new_memory:
            logger.warning("[DeepDream] No [MEMORY] section in LLM output, skipping overwrite")
            return False

        # Overwrite MEMORY.md
        try:
            main_file = self.get_main_memory_file(user_id)
            old_size = len(memory_content)
            main_file.write_text(new_memory + "\n", encoding="utf-8")
            logger.info(
                f"[DeepDream] Updated MEMORY.md "
                f"({old_size} → {len(new_memory)} chars)"
            )
        except Exception as e:
            logger.warning(f"[DeepDream] Failed to write MEMORY.md: {e}")
            return False

        # Write dream diary
        if dream_diary:
            try:
                self._write_dream_diary(dream_diary, user_id)
            except Exception as e:
                logger.warning(f"[DeepDream] Failed to write dream diary: {e}")

        logger.info("[DeepDream] ✅ Deep Dream completed successfully")
        return True

    def _read_main_memory(self, user_id: Optional[str] = None) -> str:
        """Read current MEMORY.md content."""
        main_file = self.get_main_memory_file(user_id)
        if main_file.exists():
            return main_file.read_text(encoding="utf-8").strip()
        return ""

    def _read_recent_dailies(
        self, user_id: Optional[str] = None, lookback_days: int = 1
    ) -> tuple:
        """
        Read recent daily memory files.

        Returns:
            (combined_text, has_content) tuple
        """
        from datetime import timedelta

        parts = []
        has_content = False
        today = datetime.now().date()

        for offset in range(lookback_days):
            day = today - timedelta(days=offset)
            date_str = day.strftime("%Y-%m-%d")
            if user_id:
                daily_file = self.memory_dir / "users" / user_id / f"{date_str}.md"
            else:
                daily_file = self.memory_dir / f"{date_str}.md"

            if daily_file.exists():
                content = daily_file.read_text(encoding="utf-8").strip()
                if content:
                    parts.append(f"### {date_str}\n\n{content}")
                    has_content = True
            else:
                parts.append(f"### {date_str}\n\n(no records)")

        return "\n\n".join(parts), has_content

    @staticmethod
    def _parse_dream_output(raw: str) -> tuple:
        """Parse LLM output into (new_memory, dream_diary)."""
        raw = raw.strip().replace("```", "")
        new_memory = ""
        dream_diary = ""

        if "[MEMORY]" in raw:
            start = raw.index("[MEMORY]") + len("[MEMORY]")
            end = raw.index("[DREAM]") if "[DREAM]" in raw else len(raw)
            new_memory = raw[start:end].strip()

        if "[DREAM]" in raw:
            start = raw.index("[DREAM]") + len("[DREAM]")
            dream_diary = raw[start:].strip()

        return new_memory, dream_diary

    def _write_dream_diary(self, content: str, user_id: Optional[str] = None):
        """Write dream diary to memory/dreams/YYYY-MM-DD.md."""
        dreams_dir = self.memory_dir / "dreams"
        if user_id:
            dreams_dir = self.memory_dir / "users" / user_id / "dreams"
        dreams_dir.mkdir(parents=True, exist_ok=True)

        today = datetime.now().strftime("%Y-%m-%d")
        diary_file = dreams_dir / f"{today}.md"
        diary_file.write_text(
            f"# Dream Diary: {today}\n\n{content}\n",
            encoding="utf-8",
        )
        logger.info(f"[DeepDream] Wrote dream diary to {diary_file}")

    # ---- Internal helpers ----
    
    def _summarize_messages(self, messages: List[Dict], max_messages: int = 0) -> str:
        """
        Summarize conversation messages using LLM.
        Returns empty string if LLM deems content not worth recording.
        Rule-based fallback only used when LLM call raises an exception.
        """
        conversation_text = self._format_conversation_for_summary(messages, max_messages)
        if not conversation_text.strip():
            return ""
        
        if self.llm_model:
            try:
                summary = self._call_llm_for_summary(conversation_text)
                if not _is_empty_sentinel(summary):
                    return summary.strip()
                logger.info("[MemoryFlush] LLM returned empty sentinel, skipping write")
                return ""
            except Exception as e:
                logger.warning(f"[MemoryFlush] LLM summarization failed, using fallback: {e}")
                return self._extract_summary_fallback(messages, max_messages)
        else:
            logger.info("[MemoryFlush] No LLM model available, using rule-based fallback")
            return self._extract_summary_fallback(messages, max_messages)

    def _format_conversation_for_summary(self, messages: List[Dict], max_messages: int = 0) -> str:
        """Format messages into readable conversation text for LLM summarization."""
        msgs = messages if max_messages == 0 else messages[-max_messages * 2:]
        lines = []
        for msg in msgs:
            role = msg.get("role", "")
            text = self._extract_text_from_content(msg.get("content", ""))
            if not text or not text.strip():
                continue
            text = text.strip()
            if role == "user":
                lines.append(f"用户: {text[:500]}")
            elif role == "assistant":
                lines.append(f"助手: {text[:500]}")
        return "\n".join(lines)

    @staticmethod
    def _extract_response_text(response) -> str:
        """
        Extract text from LLM response regardless of format.

        Handles:
        - Generator (MiniMax _handle_sync_response yields Claude-format dicts)
        - Claude format: {"role":"assistant","content":[{"type":"text","text":"..."}]}
        - OpenAI format: {"choices":[{"message":{"content":"..."}}]}
        - OpenAI SDK response object with .choices attribute
        """
        import types

        # Unwrap generator — consume first yielded item
        if isinstance(response, types.GeneratorType):
            try:
                response = next(response)
            except StopIteration:
                return ""

        if not response:
            return ""

        if isinstance(response, dict):
            # Check for error
            if response.get("error"):
                raise RuntimeError(response.get("message", "LLM call failed"))

            # Claude format: content is a list of blocks
            content = response.get("content")
            if isinstance(content, list):
                for block in content:
                    if isinstance(block, dict) and block.get("type") == "text":
                        return block.get("text", "")

            # OpenAI format
            choices = response.get("choices", [])
            if choices:
                return choices[0].get("message", {}).get("content", "")

        # OpenAI SDK response object
        if hasattr(response, "choices") and response.choices:
            return response.choices[0].message.content or ""

        return ""

    def _call_llm_for_summary(self, conversation_text: str) -> str:
        """Call LLM to generate a concise summary of the conversation."""
        from agent.protocol.models import LLMRequest
        
        request = LLMRequest(
            messages=[{"role": "user", "content": _summarize_user_prompt().format(conversation=conversation_text)}],
            temperature=0,
            max_tokens=500,
            stream=False,
            system=_summarize_system_prompt(),
        )
        
        response = self.llm_model.call(request)
        return self._extract_response_text(response)

    @staticmethod
    def _extract_first_meaningful_line(text: str, max_len: int = 120) -> str:
        """Extract the first meaningful line from assistant reply, skipping markdown noise."""
        import re
        for line in text.split("\n"):
            line = line.strip()
            if not line:
                continue
            # Skip markdown headings, horizontal rules, code fences, pure emoji/symbols
            if re.match(r'^(#{1,4}\s|```|---|\*\*\*|[-*]\s*$|[^\w\u4e00-\u9fff]{1,5}$)', line):
                continue
            # Strip leading markdown bold/emoji decorations
            cleaned = re.sub(r'^[\*#>\-\s]+', '', line).strip()
            cleaned = re.sub(r'^[\U0001f300-\U0001f9ff\u2600-\u27bf\s]+', '', cleaned).strip()
            if len(cleaned) >= 5:
                return cleaned[:max_len]
        return text.split("\n")[0].strip()[:max_len]

    @staticmethod
    def _extract_summary_fallback(messages: List[Dict], max_messages: int = 0) -> str:
        """
        Rule-based summary of discarded messages.
        Format: "用户问了X; 助手回答了Y" per event, compact and readable.
        """
        msgs = messages if max_messages == 0 else messages[-max_messages * 2:]

        events: List[str] = []
        current_user_text = ""
        for msg in msgs:
            role = msg.get("role", "")
            text = MemoryFlushManager._extract_text_from_content(msg.get("content", ""))
            if not text or not text.strip():
                continue
            text = text.strip()

            if role == "user":
                if len(text) <= 3:
                    continue
                current_user_text = text[:120]
            elif role == "assistant" and current_user_text:
                reply_summary = MemoryFlushManager._extract_first_meaningful_line(text)
                if reply_summary:
                    events.append(f"- 用户: {current_user_text} → 回复: {reply_summary}")
                else:
                    events.append(f"- 用户: {current_user_text}")
                current_user_text = ""

        if current_user_text:
            events.append(f"- 用户: {current_user_text}")

        return "\n".join(events[:10])
    
    @staticmethod
    def _extract_text_from_content(content) -> str:
        """Extract plain text from message content (string or content blocks)."""
        if isinstance(content, str):
            return content
        if isinstance(content, list):
            parts = []
            for block in content:
                if isinstance(block, dict) and block.get("type") == "text":
                    parts.append(block.get("text", ""))
                elif isinstance(block, str):
                    parts.append(block)
            return "\n".join(parts)
        return ""

    @classmethod
    def _strip_scheduler_pairs(cls, messages: List[Dict]) -> List[Dict]:
        """Drop scheduler-injected user/assistant pairs from a flush batch.

        A scheduler user message starts with the ``[SCHEDULED]`` marker
        (written by ``AgentBridge.remember_scheduled_output``); the message
        immediately following it (if it is an assistant turn) is its paired
        output and is dropped together. Regular user/assistant turns and
        any tool_use / tool_result blocks are preserved as-is.
        """
        if not messages:
            return messages

        SCHEDULED_PREFIX = "[SCHEDULED]"
        result = []
        skip_next_assistant = False
        for msg in messages:
            if not isinstance(msg, dict):
                result.append(msg)
                skip_next_assistant = False
                continue
            role = msg.get("role")
            if skip_next_assistant and role == "assistant":
                skip_next_assistant = False
                continue
            skip_next_assistant = False
            if role == "user":
                text = cls._extract_text_from_content(msg.get("content", ""))
                if text.lstrip().startswith(SCHEDULED_PREFIX):
                    skip_next_assistant = True
                    continue
            result.append(msg)
        return result


def create_memory_files_if_needed(workspace_dir: Path, user_id: Optional[str] = None):
    """
    Create essential memory files if they don't exist.
    Only creates MEMORY.md; daily files are created lazily on first write.
    
    Args:
        workspace_dir: Workspace directory
        user_id: Optional user ID for user-specific files
    """
    memory_dir = workspace_dir / "memory"
    memory_dir.mkdir(parents=True, exist_ok=True)
    
    # Create main MEMORY.md in workspace root (always needed for bootstrap)
    if user_id:
        user_dir = memory_dir / "users" / user_id
        user_dir.mkdir(parents=True, exist_ok=True)
        main_memory = user_dir / "MEMORY.md"
    else:
        main_memory = Path(workspace_dir) / "MEMORY.md"
    
    if not main_memory.exists():
        main_memory.write_text("")


def ensure_daily_memory_file(workspace_dir: Path, user_id: Optional[str] = None) -> Path:
    """
    Ensure today's daily memory file exists, creating it only when actually needed.
    Called lazily before first write to daily memory.
    
    Args:
        workspace_dir: Workspace directory
        user_id: Optional user ID for user-specific files
        
    Returns:
        Path to today's memory file
    """
    memory_dir = workspace_dir / "memory"
    memory_dir.mkdir(parents=True, exist_ok=True)
    
    today = datetime.now().strftime("%Y-%m-%d")
    if user_id:
        user_dir = memory_dir / "users" / user_id
        user_dir.mkdir(parents=True, exist_ok=True)
        today_memory = user_dir / f"{today}.md"
    else:
        today_memory = memory_dir / f"{today}.md"
    
    if not today_memory.exists():
        today_memory.write_text(
            f"# Daily Memory: {today}\n\n"
        )
    
    return today_memory