"""Render Anthropic ``Message`` (and user turns) into markdown. One-way: produces the human-facing artifact only. The parser strips tool/thinking callouts separately when reshaping history for backend replay. """ from __future__ import annotations import re from typing import TYPE_CHECKING, Any, cast from anthropic.types import Message, TextBlock, ThinkingBlock if TYPE_CHECKING: from anthropic.types import MessageParam __all__ = [ "FENCE", "USER_SCAFFOLD", "append_to_body", "render_assistant_message", "render_user_text", "summarize_tool_input", ] USER_SCAFFOLD = "### User:\n\n" FENCE = "````" _TITLE_KEYS = ("description", "path", "file", "filename", "url", "command", "query") def render_user_text(content: str) -> str: r"""Render one user-spoken turn as ``### User:\n\n``.""" return f"### User:\n\n{content.strip()}\n" def render_assistant_text(text: str) -> str: r"""Render a text-only assistant turn as ``### Assistant:\n\n``.""" return f"### Assistant:\n\n{text.strip()}\n" def render_assistant_message(message: Message, *, thinking_is_progress: bool) -> str: """Render an assistant ``Message`` into a turn block. Tool-use blocks render to nothing. Progress thinking renders as text in place and vanishes when empty; reasoning renders as a collapsed callout, of which only the first between two texts reaches the file. """ parts: list[str] = ["### Assistant:", ""] after_reasoning = False for block in message.content: if isinstance(block, ThinkingBlock) and not thinking_is_progress: if not after_reasoning: parts.extend((*_render_reasoning(block.thinking or ""), "")) after_reasoning = True continue text = _spoken(block) if text: parts.extend((text, "")) after_reasoning = False return "\n".join(parts).rstrip() + "\n" def render_user_param(param: MessageParam) -> str: """Render a ``MessageParam`` user message into a ``### User:`` block. Tool_result blocks in the content list are dropped silently. """ content = param.get("content", "") if isinstance(content, str): text = content else: chunks = [ str(blk.get("text", "")) for blk in content if isinstance(blk, dict) and blk.get("type") == "text" ] text = "\n\n".join(chunks) return render_user_text(text) def append_to_body(existing: str, new_block: str) -> str: """Append ``new_block`` to ``existing`` with a decorative ``---`` separator. The separator is visual only; the parser ignores it. """ head = existing.rstrip() if head: return f"{head}\n\n---\n\n{new_block}" return new_block def summarize_tool_input(name: str, tool_input: object) -> str: """Build the ``[!tool]- `` title, picking one salient input field.""" if not isinstance(tool_input, dict): return name d = cast("dict[str, Any]", tool_input) for key in _TITLE_KEYS: value = d.get(key) if isinstance(value, str) and value: short = value if len(value) <= 60 else value[:57] + "..." return f"{name} ยท {short}" return name def _render_reasoning(text: str) -> tuple[str, ...]: lines = text.strip().splitlines() or [""] return ("> [!thinking]-", *(f"> {line}" if line else ">" for line in lines)) def _spoken(block: object) -> str: if isinstance(block, TextBlock): return (block.text or "").strip() if isinstance(block, ThinkingBlock): return (block.thinking or "").strip() return "" def adaptive_fence(content: str) -> str: """Return a backtick fence longer than any backtick run in ``content``.""" longest = 0 for match in re.finditer(r"`+", content): longest = max(longest, len(match.group(0))) return "`" * max(3, longest + 1)