"""Parse a markdown chat file into Anthropic ``MessageParam`` history. The file format is documented in ``frontends/markdown/__init__.py``: ``### User:`` / ``### Assistant:`` H3 headers split turns, optional ``---`` HRs between turns are visual-only, ``> [!thinking]-`` and ``> [!tool]- `` callouts mark structured assistant content. For backend consumption we strip thinking and tool_use callouts — assistant turns become text-only. Rationale: history replay through claude-code's JSONL injection only needs the *narrated* answer (the thinking signatures expire and the original tool_results aren't captured in the renderer's output, so a faithful tool_use round-trip isn't possible today). The renderer keeps callouts in the file because they're informational for the human reader; the parser drops them when shaping the backend's input. """ from __future__ import annotations import re from dataclasses import dataclass from typing import TYPE_CHECKING, Any import frontmatter if TYPE_CHECKING: from anthropic.types import MessageParam __all__ = [ "AssistantSegment", "ParsedFile", "ParsedTurn", "TextSegment", "ToolSegment", "last_role", "parse", "parse_assistant_structure", "resolve_agent", ] @dataclass(frozen=True, slots=True) class TextSegment: """A run of plain text inside an assistant turn (between callouts).""" text: str @dataclass(frozen=True, slots=True) class ToolSegment: """A ``> [!tool]- `` callout placeholder. Only the tool ``name`` is captured — the " · summary" suffix on the callout title and the JSON body inside the quote block are decorative for the human reader; the canonical tool_use block lives in the DB and is keyed by *position+name* against the structure parsed here. """ name: str AssistantSegment = TextSegment | ToolSegment # Turn marker — must be exactly ``### User:`` or ``### Assistant:`` on # its own line. Trailing whitespace tolerated; nothing after the colon # on the same line (any inline content would mean the user typed # something that just happens to look like a header, and we'd rather # misparse than silently fold inline content into a turn). _TURN_RE = re.compile(r"^###\s+(User|Assistant):\s*$", re.MULTILINE) # Callout-start lines we strip from assistant turns when extracting # text. We don't try to parse the contents — for backend input we just # need to drop the whole quoted block. _CALLOUT_START_RE = re.compile(r"^>\s+\[!(thinking|tool)\]") # Tool-callout title line: ``> [!tool]- `` or ``> [!tool]- · ``. # We only need the ```` part for skeleton matching; the summary is # decorative (built by ``renderer.summarize_tool_input`` from inputs the # user can edit visually without semantic consequence). _TOOL_TITLE_RE = re.compile(r"^>\s+\[!tool\]-\s*(.*?)\s*$") # Renderer joins name + summary with " · " (U+00B7) — see # ``renderer.summarize_tool_input``. We split on it to recover the # bare tool name. _TOOL_TITLE_SEP = " · " @dataclass(frozen=True, slots=True) class ParsedTurn: """One turn extracted from the chat file. ``role`` is ``"user"`` or ``"assistant"``. ``text`` is the spoken content with callouts stripped and HRs dropped — used both as the backend's ``MessageParam.content`` (back-compat with the existing parser shape) and as the diff key against stored turns. ``structure`` is non-empty only for assistant turns: an ordered list of ``TextSegment`` / ``ToolSegment`` reflecting the visible layout of the assistant block, used by the conversation store to align with the canonical tool_use blocks held in DB. """ role: str text: str structure: tuple[TextSegment | ToolSegment, ...] = () @dataclass(frozen=True, slots=True) class ParsedFile: """Result of parsing a single chat ``.md``. ``metadata`` is the YAML frontmatter as a plain dict (empty if the file has none). ``messages`` is the conversation history shaped for ``Backend.complete`` — assistant turns are text-only. ``turns`` is 1:1 with ``messages`` and carries the per-turn structure (for assistant turns) that the conversation store needs to detect text-only edits vs. structural forks. ``body`` is the raw markdown content *after* the frontmatter is stripped; the renderer needs it when it appends a new assistant turn so it can preserve whatever the human typed verbatim (including any callouts or HRs they added). """ metadata: dict[str, Any] body: str messages: list[MessageParam] turns: list[ParsedTurn] def parse(text: str) -> ParsedFile: """Parse a chat ``.md`` into ``(metadata, body, messages, turns)``. A file with no turn markers but non-empty body is treated as a single user turn — the friendly path for "user types into a new file and hits send" before any turn markers exist. Assistant turns that have *only* tool callouts (no spoken text) are preserved here even though their ``MessageParam.content`` is empty — the structure carries tool-segment information the conversation store needs for skeleton matching. The renderer in practice always emits at least a trailing text block, so this branch is defensive. """ parsed = frontmatter.loads(text) metadata = dict(parsed.metadata) body = parsed.content messages: list[MessageParam] = [] parsed_turns: list[ParsedTurn] = [] raw_turns = _split_turns(body) if not raw_turns: stripped = body.strip() if stripped: messages.append({"role": "user", "content": stripped}) parsed_turns.append(ParsedTurn(role="user", text=stripped)) return ParsedFile( metadata=metadata, body=body, messages=messages, turns=parsed_turns ) for role, raw in raw_turns: if role == "user": text_content = _strip_hrs(raw).strip() if text_content: messages.append({"role": "user", "content": text_content}) parsed_turns.append(ParsedTurn(role="user", text=text_content)) else: structure = parse_assistant_structure(raw) text_content = _segments_to_spoken_text(structure) has_tools = any(isinstance(s, ToolSegment) for s in structure) if text_content: messages.append({"role": "assistant", "content": text_content}) parsed_turns.append( ParsedTurn( role="assistant", text=text_content, structure=tuple(structure) ) ) elif has_tools: # Tool-only assistant turn: nothing to feed the backend # as ``content`` (it'd reject an empty string), but the # structure must survive so the store can align it # against stored tool_use blocks. We synthesize a # single-space text content for backend round-trip; the # conversation store will replace this payload with the # canonical stored blocks before the backend ever sees # it on a continuation. messages.append({"role": "assistant", "content": " "}) parsed_turns.append( ParsedTurn(role="assistant", text="", structure=tuple(structure)) ) return ParsedFile( metadata=metadata, body=body, messages=messages, turns=parsed_turns ) def parse_assistant_structure(raw: str) -> list[TextSegment | ToolSegment]: """Walk an assistant turn body, return its ordered text/tool segments. Tool callouts become :class:`ToolSegment` with just the tool name — the title's optional ``" · summary"`` suffix and the JSON body inside the quote block are decorative; the canonical tool_use block is held in the conversation store. Thinking callouts are stripped entirely (they were never round-trippable through the file — signatures expire). HR separator lines drop out. Empty / whitespace-only text segments at the boundaries (start, end, between adjacent tool callouts) are dropped so the skeleton is robust against renderer whitespace choices; a non-empty text segment with surrounding whitespace is trimmed on both ends but preserved. """ segments: list[TextSegment | ToolSegment] = [] pending_text: list[str] = [] def _flush_text() -> None: if not pending_text: return joined = "\n".join(pending_text) # Collapse runs of >2 blank lines (created when we stripped a # mid-block callout) into one so the diff against a re-render # is stable. cleaned = re.sub(r"\n{3,}", "\n\n", joined).strip() pending_text.clear() if cleaned: segments.append(TextSegment(text=cleaned)) lines = raw.splitlines() i = 0 while i < len(lines): line = lines[i] callout_match = _CALLOUT_START_RE.match(line) if callout_match: kind = callout_match.group(1) # Capture tool name *before* advancing past the block. if kind == "tool": title_match = _TOOL_TITLE_RE.match(line) title = title_match.group(1) if title_match else "" name = title.split(_TOOL_TITLE_SEP, 1)[0].strip() _flush_text() segments.append(ToolSegment(name=name)) else: # Thinking callout — drop the whole block, emit nothing. _flush_text() # Skip the rest of the quote block. while i < len(lines) and lines[i].lstrip().startswith(">"): i += 1 continue if line.strip() == "---": i += 1 continue pending_text.append(line) i += 1 _flush_text() return segments def _segments_to_spoken_text(segments: list[TextSegment | ToolSegment]) -> str: r"""Reduce a structure list to the spoken-text view the backend sees. Concatenates :class:`TextSegment` contents with ``\n\n`` between them, dropping :class:`ToolSegment` entries. Equivalent to what the pre-Conversation-store parser did — we keep that behavior so existing fingerprints (frontmatter ``fingerprint`` field) stay valid. """ chunks = [s.text for s in segments if isinstance(s, TextSegment)] return "\n\n".join(c for c in chunks if c).strip() def last_role(messages: list[MessageParam]) -> str | None: """Return ``"user"`` / ``"assistant"`` / ``None`` for an empty list.""" if not messages: return None return messages[-1]["role"] def resolve_agent( *, metadata: dict[str, Any], request_override: str | None, default: str | None ) -> str | None: """Resolve the agent for this chat. Precedence: request body override > frontmatter > frontend default. Returns ``None`` if none match — caller responds with 400. """ if request_override: return request_override fm_agent = metadata.get("agent") if isinstance(fm_agent, str) and fm_agent: return fm_agent return default # ---- internals --------------------------------------------------------- def _split_turns(body: str) -> list[tuple[str, str]]: """Walk turn markers, return ``[(role_lc, raw_body), ...]``. Body for each turn is everything between this marker and the next (or EOF). Leading marker line itself is dropped. We don't trim whitespace here — that's per-role. """ matches = list(_TURN_RE.finditer(body)) if not matches: return [] out: list[tuple[str, str]] = [] for i, m in enumerate(matches): role = m.group(1).lower() start = m.end() end = matches[i + 1].start() if i + 1 < len(matches) else len(body) out.append((role, body[start:end])) return out def _strip_hrs(raw: str) -> str: """Drop decorative ``---`` separator lines (whole-line HRs only). A ``---`` mid-paragraph (rare, but possible) stays. Only lines that are *exactly* the HR after optional surrounding whitespace are removed — those are the ones the renderer emits between turns. """ lines = raw.splitlines() kept = [ln for ln in lines if ln.strip() != "---"] return "\n".join(kept)