prompt.py

# 修改历史 (Revision History)
# ==================================
# 版本: v1.1
# 日期: 2026-06-24
# 修改说明: 重构 messages_to_prompt 以引入智能截断与清洗逻辑（上限 3800 字符），当 System Prompt 超过 1500 字符时自动裁切，对话历史按时间从新到旧截断，从而彻底规避 Agent 助手（如 Claude Code）注入超长提示词导致的 upstream text-too-long 报错。

"""Flatten an OpenAI ``messages`` array into a single Copilot prompt.

Copilot's protocol has no role/system channel — it takes one prompt string per
turn — so we collapse the whole conversation into one piece of text.
"""

from typing import Any, List, Optional, Union

from .schemas import ChatMessage


def content_text(content: Optional[Union[str, List[Any]]]) -> str:
    """Extract plain text from a message's content (string or content-parts)."""
    if content is None:
        return ""
    if isinstance(content, str):
        return content
    parts = []
    for part in content:
        if isinstance(part, dict):
            if part.get("type") == "text":
                parts.append(part.get("text", ""))
        else:
            parts.append(str(part))
    return "\n".join(p for p in parts if p)


def messages_to_prompt(messages: List[ChatMessage], max_len: int = 3800) -> str:
    """Flatten an OpenAI ``messages`` array into a single Copilot prompt with smart truncation."""
    # 1. 提取并初步拼接 system 消息
    system_parts = [content_text(m.content) for m in messages if m.role == "system" and m.content]
    system = "\n\n".join(system_parts)
    
    # 若 system 提示词过长，进行合理折中剪裁
    max_sys_len = 1500
    if len(system) > max_sys_len:
        keep = max_sys_len // 2 - 50
        system = system[:keep] + "\n...[System prompt truncated due to length limit]...\n" + system[-keep:]
        
    # 2. 提取对话历史消息，采用倒序追加方式控制长度在 max_len - len(system) 之内
    convo = [m for m in messages if m.role != "system"]
    
    body_parts = []
    current_body_len = 0
    
    # 倒序遍历（从新到旧），以保障最近的对话上下文和用户最新的问题不被舍弃
    for m in reversed(convo):
        label = "User" if m.role == "user" else "Assistant"
        line = f"{label}: {content_text(m.content)}"
        line_len = len(line) + 1  # 预估换行符开销
        
        allowed_body_len = max_len - len(system) - 15  # 预留一定裕量
        if current_body_len + line_len > allowed_body_len:
            # 如果这是用户最新发出的请求（也就是倒序中第一条），即使它自己也超长，我们也必须截断它本身保留核心部分
            if not body_parts:
                keep = allowed_body_len - 100
                if keep > 200:
                    truncated_content = content_text(m.content)[:keep] + "\n...[User prompt truncated]..."
                    body_parts.append(f"{label}: {truncated_content}")
            break
        body_parts.append(line)
        current_body_len += line_len
        
    # 恢复正序
    body_parts.reverse()
    
    # 补上 cue Copilot continue 标志
    if len(convo) > 1 or (len(convo) == 1 and convo[0].role != "user"):
        body_parts.append("Assistant:")
        
    body = "\n".join(body_parts)
    
    # 3. 最终拼接
    if system and body:
        return f"{system}\n\n{body}"
    return system or body