Class TranscriptWriter

src/agent/transcript.py:157–235

Append-only writer for the JSONL transcript at ``path``. Opens an ``O_APPEND`` file descriptor on construction. Each ``append(message)`` call serializes one line and emits it via a single ``os.write`` so concurrent writers (e.g. the same agent across crash-restart, or a future multi

class TranscriptWriter:
    """Append-only writer for the JSONL transcript at ``path``.

    Opens an ``O_APPEND`` file descriptor on construction. Each
    ``append(message)`` call serializes one line and emits it via a
    single ``os.write`` so concurrent writers (e.g. the same agent
    across crash-restart, or a future multi-writer scenario) cannot
    interleave bytes at sub-PIPE_BUF line sizes.

    The writer is **synchronous** (file IO, not asyncio). Per the A6/C5
    contract, callers must NOT hold ``RuntimeTaskRegistry``'s RLock
    across an ``append()`` call: while file IO is fast, blocking under
    the registry's lock would deadlock the asyncio scheduler against
    bash worker threads.
    """

    def __init__(self, path: str | Path) -> None:
        self._path = str(path)
        # Ensure parent dir exists (transcript root may not have been
        # created yet if the caller bypassed ``get_agent_transcript_path``).
        Path(self._path).parent.mkdir(parents=True, exist_ok=True)
        # ``O_APPEND`` makes every write atomic at the file-position
        # level. ``O_CLOEXEC`` keeps the fd from leaking to bash
        # subprocesses. ``0o600`` because transcripts can contain
        # sensitive prompt content — readable by the user only.
        self._fd: int | None = os.open(
            self._path,
            os.O_WRONLY | os.O_APPEND | os.O_CREAT | os.O_CLOEXEC,
            0o600,
        )
        self._closed = False

    @property
    def path(self) -> str:
        return self._path

    def append(self, message: Any) -> None:
        """Append one message as a UTF-8 JSON line, terminated with ``\\n``.

        Crash-safety: a single ``os.write`` of the line ensures the
        writer either appends the whole line or nothing. POSIX
        guarantees this for sizes ≤ ``PIPE_BUF`` (≥4096 bytes on every
        modern Unix); for larger lines the reader is tolerant of
        partial trailing content per the JSONL format design.
        """
        if self._closed or self._fd is None:
            raise RuntimeError("TranscriptWriter is closed")
        line = _serialize_message(message) + "\n"
        encoded = line.encode("utf-8")
        # ``os.write`` may short-write under specific OS conditions;
        # loop until the whole buffer is on disk. For O_APPEND files
        # the returned ``n`` is byte count of THIS write, so the loop
        # is straightforward.
        view = memoryview(encoded)
        while view:
            written = os.write(self._fd, view)
            if written <= 0:
                # Defensive: 0-byte returns shouldn't happen on regular