325 lines
12 KiB
Python
325 lines
12 KiB
Python
"""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]- <name>`` 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]- <name>`` 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]- <name>`` or ``> [!tool]- <name> · <summary>``.
|
|
# We only need the ``<name>`` 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)
|