97 lines
2.9 KiB
Python
97 lines
2.9 KiB
Python
"""Cross-frontend turn record.
|
|
|
|
Frontends emit a :class:`TurnRecord` to every handler in
|
|
``GatewayRuntime.turn_log_handlers`` after a turn finishes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
if TYPE_CHECKING:
|
|
from anthropic.types import Message, MessageParam
|
|
|
|
|
|
__all__ = ["TurnRecord", "slugify"]
|
|
|
|
|
|
_SLUG_BAD_RE = re.compile(r"[^\w\s\-]+", flags=re.UNICODE)
|
|
_SLUG_SEP_RE = re.compile(r"[\s\-]+", flags=re.UNICODE)
|
|
|
|
|
|
def slugify(text: str, *, maxlen: int = 40) -> str:
|
|
"""Sanitize ``text`` for use as a filename fragment.
|
|
|
|
Truncates to ``maxlen`` and returns ``"untitled"`` for empty input.
|
|
"""
|
|
cleaned = _SLUG_BAD_RE.sub(" ", text).strip()
|
|
cleaned = _SLUG_SEP_RE.sub("-", cleaned).strip("-")
|
|
if not cleaned:
|
|
return "untitled"
|
|
if len(cleaned) > maxlen:
|
|
cleaned = cleaned[:maxlen].rstrip("-")
|
|
return cleaned or "untitled"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class TurnRecord:
|
|
"""One completed turn, as seen by a frontend.
|
|
|
|
``input_messages`` excludes the assistant reply; ``source`` names the
|
|
producing frontend so a cross-frontend logger can skip its own turns.
|
|
"""
|
|
|
|
agent_name: str
|
|
input_messages: list[MessageParam]
|
|
output_message: Message
|
|
thinking_is_progress: bool
|
|
system: str | None = None
|
|
source: str = "anthropic"
|
|
|
|
@property
|
|
def first_user_text(self) -> str:
|
|
"""Empty string if the input history has no user turn."""
|
|
for msg in self.input_messages:
|
|
if msg.get("role") == "user":
|
|
return _text_of(msg.get("content", ""))
|
|
return ""
|
|
|
|
@property
|
|
def last_user_text(self) -> str:
|
|
for msg in reversed(self.input_messages):
|
|
if msg.get("role") == "user":
|
|
return _text_of(msg.get("content", ""))
|
|
return ""
|
|
|
|
@property
|
|
def assistant_text(self) -> str:
|
|
"""Plain text of the assistant reply, with thinking/tool blocks dropped."""
|
|
chunks = [
|
|
getattr(block, "text", "") or ""
|
|
for block in getattr(self.output_message, "content", ())
|
|
if getattr(block, "type", None) == "text"
|
|
]
|
|
return "\n\n".join(c for c in chunks if c)
|
|
|
|
|
|
def _text_of(content: object) -> str:
|
|
"""Flatten an Anthropic ``MessageParam.content`` field to a plain string.
|
|
|
|
Handles both raw-string and block-list shapes; unrecognized content is
|
|
silently skipped since this exists for naming, not faithful reconstruction.
|
|
"""
|
|
if isinstance(content, str):
|
|
return content
|
|
if isinstance(content, list):
|
|
parts: list[str] = []
|
|
for blk in content:
|
|
if not isinstance(blk, dict):
|
|
continue
|
|
d: Any = blk
|
|
if d.get("type") == "text":
|
|
parts.append(str(d.get("text", "")))
|
|
return "\n\n".join(p for p in parts if p)
|
|
return ""
|