interactive_chat_projection.py
Rama main · réplica de lectura del repositorio original.
"""Incremental, bounded text-only chat projection; no server imports or timers."""
from __future__ import annotations
import json
import os
from pathlib import Path
import stat
import threading
from typing import Callable
from public_chat_transcript import (
MAX_LINE_BYTES, MAX_LOG_BYTES, MAX_MESSAGE_BYTES, MAX_MESSAGES,
MAX_RECORDS, MAX_TEXT_BYTES, PublicTranscriptError,
_is_goal_followup, _public_timestamp,
)
class IncrementalChatTranscript:
"""Use one fresh privacy projector for the lifetime of an append-only log.
The owner must discard this reader after a known history rewrite or privacy
projection change. Stat checks detect replacement, truncation and same-size
changes; no incremental reader can prove an earlier prefix was not changed
in place while also growing without rereading that prefix.
"""
def __init__(self, path: Path, projector: Callable[[dict], dict | None]):
self.path = Path(path)
self.projector = projector
self._lock = threading.Lock()
self._stamp = None
self._offset = 0
self._revision = 0
self._records = 0
self._total_text = 0
self._messages: list[dict] = []
self._outputs: dict[str, list[str]] = {}
self._failed = False
def _snapshot(self) -> dict:
return {"messages": [dict(message) for message in self._messages],
"through_bytes": self._offset, "revision": str(self._revision)}
@staticmethod
def _file_stamp(value):
return (value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns, value.st_ctime_ns)
def load(self) -> dict:
"""Read only new complete records; a detected error requires a new reader."""
with self._lock:
if self._failed:
raise PublicTranscriptError("Chat history changed; reopen the conversation")
try:
return self._load()
except Exception as exc:
# The supplied projector has private state and cannot be rolled
# back after a partially consumed bad page. Never reuse it.
self._failed = True
if isinstance(exc, PublicTranscriptError):
raise
raise PublicTranscriptError("Chat history is unavailable") from exc
def _load(self) -> dict:
try:
fd = os.open(self.path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
except FileNotFoundError:
if self._stamp is None:
return self._snapshot()
raise PublicTranscriptError("Chat history changed; reopen the conversation") from None
with os.fdopen(fd, "rb") as stream:
initial = os.fstat(stream.fileno())
if not stat.S_ISREG(initial.st_mode):
raise PublicTranscriptError("Chat history is unavailable")
stamp = self._file_stamp(initial)
if initial.st_size > MAX_LOG_BYTES:
raise PublicTranscriptError("Chat exceeds the 64 MiB sharing limit")
if self._stamp is not None:
if (stamp[:2] != self._stamp[:2] or stamp[2] < self._stamp[2]
or stamp[2] == self._stamp[2] and stamp[3:] != self._stamp[3:]):
raise PublicTranscriptError("Chat history changed; reopen the conversation")
if stamp == self._stamp:
return self._snapshot()
stream.seek(self._offset)
while self._offset < initial.st_size:
line = stream.readline(min(MAX_LINE_BYTES + 1, initial.st_size - self._offset))
if len(line) > MAX_LINE_BYTES:
raise PublicTranscriptError("A chat record exceeds the sharing limit")
if not line.endswith(b"\n"):
break # Re-read this uncommitted tail only after it grows.
self._offset += len(line)
self._records += 1
if self._records > MAX_RECORDS:
raise PublicTranscriptError("Chat exceeds the sharing record limit")
try:
raw = json.loads(line)
except (ValueError, UnicodeError, RecursionError) as exc:
raise PublicTranscriptError("Chat history contains an unreadable record") from exc
if not isinstance(raw, dict) or not isinstance(raw.get("type"), str):
raise PublicTranscriptError("Chat history contains an invalid event")
self._consume(raw)
final = os.fstat(stream.fileno())
current = self.path.stat(follow_symlinks=False)
if (not stat.S_ISREG(current.st_mode)
or (current.st_dev, current.st_ino) != stamp[:2]
or final.st_size < initial.st_size
or final.st_size == initial.st_size and self._file_stamp(final)[3:] != stamp[3:]):
raise PublicTranscriptError("Chat history changed; reopen the conversation")
# Save the boundary actually read, not a later concurrent append.
self._stamp = stamp
return self._snapshot()
def _consume(self, raw: dict) -> None:
kind = raw["type"]
if kind not in {"turn_started", "assistant_text", "turn_finished", "reasoning_summary"} and not _is_goal_followup(raw):
return
if kind == "reasoning_summary" and raw.get("phase") != "commentary":
return
event = self.projector(raw)
if not event:
return
if not isinstance(event, dict):
raise PublicTranscriptError("Chat history projection is invalid")
run = str(event.get("run_id") or "")
if kind == "turn_started" or _is_goal_followup(event):
self._outputs[run] = []
role, text = "user", event.get("prompt")
else:
role = "assistant"
text = event.get("result_text") if kind == "turn_finished" else event.get("text")
if not isinstance(text, str) or not text.strip():
return
try:
size = len(text.encode("utf-8"))
except UnicodeEncodeError as exc:
raise PublicTranscriptError("Chat history contains invalid Unicode text") from exc
if size > MAX_MESSAGE_BYTES:
raise PublicTranscriptError("A message exceeds the 256 KiB sharing limit")
normalized = " ".join(text.split())
previous = self._outputs.setdefault(run, [])
if kind == "turn_finished" and (normalized in previous or normalized == " ".join(previous)):
return
if kind == "assistant_text":
previous.append(normalized)
self._total_text += size
if self._total_text > MAX_TEXT_BYTES or len(self._messages) >= MAX_MESSAGES:
raise PublicTranscriptError("Chat exceeds the 1,000-message / 2 MiB sharing limit")
message = {"role": role, "text": text}
timestamp = _public_timestamp(event.get("ts"))
if timestamp is not None:
message["timestamp"] = timestamp
self._messages.append(message)
self._revision = self._offset