def read_file(path: str, start_line: int = 1, limit: int = 200) -> str:
    try:
        target = safe_path(path)
        if not target.exists():
            return f"Error: file does not exist - {path}"
        if not target.is_file():
            return f"Error: path is not a file - {path}"

        lines = target.read_text(encoding="utf-8").splitlines()
        start = max(1, start_line)
        end = min(len(lines), start + limit - 1)

        selected = lines[start - 1 : end]
        rendered = [f"{line_no}: {line}" for line_no, line in enumerate(selected, start)]

        if end < len(lines):
            rendered.append(f"... ({len(lines) - end} more lines)")

        return "\n".join(rendered)[:50000]
    except Exception as exc:
        return f"Error: {exc}"

MAX_TOOL_OUTPUT_CHARS = 50000


def trim_tool_output(output: str) -> str:
    if len(output) <= MAX_TOOL_OUTPUT_CHARS:
        return output

    head = output[:20000]
    tail = output[-8000:]
    omitted = len(output) - len(head) - len(tail)

    return f"""
{head}

... ({omitted} characters omitted) ...

{tail}
""".strip()

import hashlib
import json
from dataclasses import dataclass, field
from typing import Any


@dataclass
class ContextState:
    summary: str = ""
    tool_artifacts: dict[str, str] = field(default_factory=dict)
    token_budget: int = 24000
    recent_turns_to_keep: int = 6

def compact_tool_message(
    message: dict[str, Any],
    state: ContextState,
    limit: int = 3000,
) -> dict[str, Any]:
    if message.get("role") != "tool":
        return message

    content = str(message.get("content") or "")
    if len(content) <= limit:
        return message

    digest = hashlib.sha256(content.encode("utf-8")).hexdigest()[:12]
    artifact_id = f"tool-result-{digest}"
    state.tool_artifacts[artifact_id] = content

    compacted = f"""
[large tool result compacted]
artifact_id: {artifact_id}
original_chars: {len(content)}
sha256: {digest}

head:
{content[:1200]}

tail:
{content[-800:]}
""".strip()

    return {**message, "content": compacted}

def read_tool_artifact(
    state: ContextState,
    artifact_id: str,
    start: int = 0,
    limit: int = 3000,
) -> str:
    content = state.tool_artifacts.get(artifact_id)
    if content is None:
        return f"Error: artifact not found - {artifact_id}"

    start = max(0, start)
    limit = max(1, min(limit, 10000))
    chunk = content[start : start + limit]
    next_offset = start + len(chunk)
    suffix = f"\n[next_offset={next_offset}]" if next_offset < len(content) else ""
    return chunk + suffix

def estimate_tokens(text: str) -> int:
    ascii_chars = sum(1 for char in text if ord(char) < 128)
    non_ascii_chars = len(text) - ascii_chars
    return max(1, ascii_chars // 4 + non_ascii_chars)


def message_to_text(message: dict[str, Any]) -> str:
    parts = [
        f"role: {message.get('role')}",
        f"content: {message.get('content') or ''}",
    ]

    if message.get("tool_calls"):
        parts.append(f"tool_calls: {message['tool_calls']}")

    return "\n".join(parts)


def estimate_messages_tokens(messages: list[dict[str, Any]]) -> int:
    return sum(estimate_tokens(message_to_text(message)) for message in messages)

context_state = ContextState(
    token_budget=24000,
    recent_turns_to_keep=6,
)

SUMMARY_PROMPT = """
你负责压缩 Agent Harness 的历史上下文。

请保留：
1. 用户当前目标和明确要求；
2. 已经完成的重要步骤；
3. 当前仍然有效的约束；
4. 修改过、读取过、需要继续关注的文件；
5. 工具调用得到的关键结论；
6. 失败过的命令、错误原因和已排除路径；
7. 当前 Todo 状态和下一步建议。

请丢弃：
1. 大段原始工具输出；
2. 重复日志；
3. 已经失效的中间尝试；
4. 和当前任务无关的细节。
"""


def update_context_summary(
    state: ContextState,
    old_messages: list[dict[str, Any]],
) -> None:
    payload = {
        "existing_summary": state.summary,
        "old_messages": old_messages,
    }

    response = client.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": SUMMARY_PROMPT},
            {
                "role": "user",
                "content": json.dumps(payload, ensure_ascii=False, indent=2),
            },
        ],
        max_completion_tokens=2000,
    )

    choice = response.choices[0]
    if choice.finish_reason == "length":
        raise RuntimeError("context summary was truncated")

    summary = (choice.message.content or "").strip()
    if not summary:
        raise RuntimeError("context summary is empty")
    state.summary = summary[:8000]

class MessageProtocolError(ValueError):
    pass


def group_protocol_turns(
    messages: list[dict[str, Any]],
) -> list[list[dict[str, Any]]]:
    """Keep an assistant tool call and all of its tool results together."""
    groups: list[list[dict[str, Any]]] = []
    index = 0

    while index < len(messages):
        message = messages[index]
        tool_calls = message.get("tool_calls") or []

        if message.get("role") == "assistant" and tool_calls:
            expected_ids = [str(call.get("id") or "") for call in tool_calls]
            if any(not item for item in expected_ids) or len(set(expected_ids)) != len(
                expected_ids
            ):
                raise MessageProtocolError("assistant tool calls need unique ids")

            group = [message]
            seen_ids: set[str] = set()
            index += 1

            while index < len(messages) and messages[index].get("role") == "tool":
                tool_message = messages[index]
                tool_call_id = str(tool_message.get("tool_call_id") or "")
                if tool_call_id not in expected_ids or tool_call_id in seen_ids:
                    raise MessageProtocolError(
                        f"unexpected tool result: {tool_call_id or 'missing id'}"
                    )
                group.append(tool_message)
                seen_ids.add(tool_call_id)
                index += 1

            if seen_ids != set(expected_ids):
                missing = sorted(set(expected_ids) - seen_ids)
                raise MessageProtocolError(f"missing tool results: {missing}")

            groups.append(group)
            continue

        if message.get("role") == "tool":
            raise MessageProtocolError("tool result has no preceding assistant call")

        groups.append([message])
        index += 1

    return groups


def split_for_compaction(
    messages: list[dict[str, Any]],
    keep_recent_turns: int,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    groups = group_protocol_turns(messages)
    if len(groups) <= keep_recent_turns:
        return [], messages

    cutoff = len(groups) - keep_recent_turns
    old_messages = [message for group in groups[:cutoff] for message in group]
    recent_messages = [message for group in groups[cutoff:] for message in group]
    return old_messages, recent_messages


def compact_messages_if_needed(
    messages: list[dict[str, Any]],
    state: ContextState,
) -> list[dict[str, Any]]:
    # 结果压缩和历史裁剪都写回 canonical messages，避免下一轮重复处理旧历史。
    messages[:] = [compact_tool_message(message, state) for message in messages]

    summary_tokens = estimate_tokens(state.summary) if state.summary else 0
    if estimate_messages_tokens(messages) + summary_tokens <= state.token_budget:
        return messages

    old_messages, recent_messages = split_for_compaction(
        messages,
        keep_recent_turns=state.recent_turns_to_keep,
    )

    if old_messages:
        update_context_summary(state, old_messages)
        messages[:] = recent_messages

    return messages

class ContextManager:
    def __init__(self, state: ContextState):
        self.state = state

    @property
    def summary(self) -> str:
        return self.state.summary

    def compact(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
        return compact_messages_if_needed(messages, self.state)

    def hard_compact(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
        messages[:] = hard_compact_messages(messages, self.state)
        return messages

def build_model_messages(
    messages: list[dict[str, Any]],
    context_manager: ContextManager,
) -> list[dict[str, Any]]:
    # compact() 可能刚刚更新 summary，所以必须先压缩、后读取摘要。
    working_messages = context_manager.compact(messages)

    model_messages = [{"role": "system", "content": SYSTEM_PROMPT}]

    if context_manager.summary:
        model_messages.append(
            {
                "role": "system",
                "content": (
                    "下面是被压缩的历史上下文摘要：\n"
                    + context_manager.summary
                ),
            }
        )

    model_messages.extend(working_messages)
    return model_messages

def run_loop(user_message: str, max_steps: int = 8) -> str:
    messages: list[dict[str, Any]] = [{"role": "user", "content": user_message}]
    context_manager = ContextManager(
        ContextState(token_budget=24000, recent_turns_to_keep=6)
    )

    for step in range(max_steps):
        response = client.chat.completions.create(
            model=MODEL,
            messages=build_model_messages(messages, context_manager),
            tools=TOOLS,
        )

        # 后面的工具调用处理逻辑继续沿用前几篇

def hard_compact_messages(
    messages: list[dict[str, Any]],
    state: ContextState,
) -> list[dict[str, Any]]:
    if not messages:
        return []

    compacted = [compact_tool_message(message, state) for message in messages]
    old_messages, recent_messages = split_for_compaction(
        compacted,
        keep_recent_turns=3,
    )
    if old_messages:
        update_context_summary(state, old_messages)

    latest_user = next(
        (message for message in reversed(compacted) if message.get("role") == "user"),
        None,
    )

    if latest_user and all(message is not latest_user for message in recent_messages):
        return [latest_user] + recent_messages

    return recent_messages