event_writer.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/event_writer.py
 
"""
NDJSON event-log writer — M2.2 (event substrate).
 
Purpose
-------
Persist every ``option_c_agent_contracts.TurnEvent`` published on the M2.1
``EventBus`` to disk, one canonical-JSON line at a time. The writer subscribes
to the bus as a wildcard (``"*"``) consumer and is the first real exerciser of
the M2.1 surface (plan 098). Three disciplines load-bear:
 
- **Atomic** — each ``_on_event`` writes one full line via a single ``write()``
  + ``fsync()`` (the file is opened unbuffered), so a crash mid-call leaves
  either a complete prior line or an incomplete trailing line — never a torn
  middle line that breaks NDJSON parse of earlier events.
- **Locked** — cross-process exclusion on the active file via a stdlib-only
  ``_FileLock`` (``fcntl`` on POSIX, ``msvcrt`` on Windows). No third-party
  ``filelock`` / ``portalocker`` dependency (ADR-1).
- **Rotating** — when the active file reaches ``max_file_bytes`` the writer
  rotates in-line at write time (never on a background thread, ADR-3): the
  triggering event is written to the current file, then the next sequence slot
  is opened for subsequent events.
 
Replay tools (M2.7) parse the lines back with ``TurnEvent.model_validate_json``.
 
Responsibilities
----------------
- Expose ``EventWriter`` with ``attach`` / ``detach`` / ``close`` / ``_on_event``.
- Serialize via ``canonical_json(event.model_dump(mode="json"))``.
- Resume from the highest existing rotation slot in ``base_dir`` (ADR-4).
- Surface its own write/serialize failures via ``_dbg`` (L2) before re-raising;
  never silently swallow (the bus isolates the exception per M2.1 ADR-3).
 
Out of scope (deferred): ``run_turn()`` integration (M2.3), session-identity
binding (M2.5), replay (M2.7), concurrent-writer stress (M2.8), async/queued
writes (synchronous per M2.1 ADR-1).
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — write / serialize / fsync failures (surfaced before re-raise)
    L3 — attach / detach / rotation lifecycle
    L4 — module-load coverage marker
 
Contracts
---------
- One ``write()`` + one ``fsync()`` per published event; the file is unbuffered.
- Rotation is observed synchronously after each write (ADR-3); no sidecar thread.
- ``EventWriter`` adds no third-party dependency — locking is stdlib only.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.event_bus (EventBus / Subscription)
- oc_chat.core.events (TurnEvent; runs the contracts-import bootstrap)
- option_c_agent_contracts (canonical_json serialisation of each event line)
"""
from __future__ import annotations
 
import os
import re
import sys
import threading
from dataclasses import dataclass, field
from pathlib import Path
from typing import IO, Optional
 
from oc_diagnostics import _dbg
 
# Importing the bus pulls in oc_chat.core.events, whose module-level bootstrap
# makes option_c_agent_contracts importable before the canonical_json import.
from oc_chat.core.event_bus import EventBus, Subscription
from oc_chat.core.events import TurnEvent
from option_c_agent_contracts import canonical_json
 
_dbg("OC_CHAT.MAIN", "event_writer", "module", 4)
 
 
class _FileLock:
    """Exclusive, cross-platform advisory lock over an open binary file.
 
    Stdlib only (ADR-1): ``fcntl.flock`` on POSIX, ``msvcrt.locking`` on Windows
    (locking a single byte at offset 0, a region both writer processes agree
    on). Used as a context manager around the write + fsync of one line.
    """
 
    def __init__(self, fileobj: IO[bytes]) -> None:
        self._fileobj = fileobj
 
    def __enter__(self) -> "_FileLock":
        self.acquire()
        return self
 
    def __exit__(self, *_exc: object) -> None:
        self.release()
 
    def acquire(self) -> None:
        func = "_FileLock.acquire"
        fd = self._fileobj.fileno()
        if sys.platform == "win32":
            import msvcrt
 
            saved = self._fileobj.tell()
            self._fileobj.seek(0)
            try:
                msvcrt.locking(fd, msvcrt.LK_LOCK, 1)
            finally:
                self._fileobj.seek(saved)
        else:
            import fcntl
 
            fcntl.flock(fd, fcntl.LOCK_EX)
        _dbg("OC_CHAT.MAIN", func, "acquired", 4)
 
    def release(self) -> None:
        func = "_FileLock.release"
        fd = self._fileobj.fileno()
        if sys.platform == "win32":
            import msvcrt
 
            saved = self._fileobj.tell()
            self._fileobj.seek(0)
            try:
                msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
            finally:
                self._fileobj.seek(saved)
        else:
            import fcntl
 
            fcntl.flock(fd, fcntl.LOCK_UN)
        _dbg("OC_CHAT.MAIN", func, "released", 4)
 
 
def _missing_ancestors(target: Path) -> tuple[Path, ...]:
    """Return the *target* chain that does not yet exist, DEEPEST FIRST.
 
    Used to build an ownership receipt before ``mkdir(parents=True)``: every path
    returned here is one this attempt is about to create, so an abort may remove
    exactly those (and nothing above them). Deepest-first is the safe removal
    order.
    """
    missing: list[Path] = []
    node = Path(target)
    while not node.exists():
        missing.append(node)
        parent = node.parent
        if parent == node:          # filesystem root — stop, never walk past it
            break
        node = parent
    return tuple(missing)
 
 
@dataclass(frozen=True)
class EventWriterOwnershipReceipt:
    """What a PREPARE attempt created, so an abort can undo exactly that (R09-07).
 
    The distinction is load-bearing: an aborting transition must leave the
    filesystem byte-identical to the pre-prepare snapshot, which means removing
    the directories and the slot file this attempt made while never deleting a
    pre-existing event log that merely happened to be in the way.
    """
 
    #: Directories this attempt created, deepest first. Removed only if empty.
    created_dirs: tuple[Path, ...] = field(default_factory=tuple)
    #: The slot file this attempt created, if any. Removed only if still empty.
    created_file: Optional[Path] = None
    #: ``True`` when the slot file was already on disk (never removed).
    file_preexisted: bool = False
 
 
@dataclass
class PreparedEventWriter:
    """An opened-but-unsubscribed :class:`EventWriter` plus its ownership receipt.
 
    Produced by :meth:`EventWriter.open_unsubscribed` in a transition's PREPARE
    phase. Either the transition commits — and
    :func:`oc_chat.core.event_bootstrap.install_prepared_writer` performs the
    linearizable bus swap — or it aborts and :meth:`abort` undoes this attempt's
    filesystem effects.
 
    :meth:`abort` is IDEMPOTENT and safe to call repeatedly, including from a
    resource-owning context manager unwinding under a repeated ``KeyboardInterrupt``.
    """
 
    writer: "EventWriter"
    receipt: EventWriterOwnershipReceipt
    _aborted: bool = field(default=False, repr=False)
 
    def abort(self) -> None:
        """Close the descriptor and remove ONLY what this attempt created.
 
        Every step is individually guarded: a failure to remove one artifact must
        not prevent the rest of the cleanup, and cleanup must never raise into an
        already-failing unwind path. A created file is removed only while it is
        still empty (size 0) — if anything was written to it, it is real data and
        is kept.
        """
        func = "PreparedEventWriter.abort"
        if self._aborted:
            return                                  # idempotent — repeated SIGINT safe
        self._aborted = True
        try:
            self.writer.close()
        except Exception as exc:                    # pragma: no cover - defensive
            _dbg("OC_CHAT.MAIN", func, f"EXTERNAL_ERROR: close failed: {exc}", 2)
        created = self.receipt.created_file
        if created is not None:
            try:
                if created.exists() and created.stat().st_size == 0:
                    created.unlink()
                elif created.exists():
                    _dbg("OC_CHAT.MAIN", func, f"kept non-empty {created.name} (has data)", 2)
            except Exception as exc:
                _dbg("OC_CHAT.MAIN", func, f"EXTERNAL_ERROR: unlink failed: {exc}", 2)
        for node in self.receipt.created_dirs:      # deepest first
            try:
                if node.is_dir() and not any(node.iterdir()):
                    node.rmdir()
            except Exception as exc:
                _dbg("OC_CHAT.MAIN", func, f"EXTERNAL_ERROR: rmdir {node} failed: {exc}", 2)
        _dbg("OC_CHAT.MAIN", func, "aborted prepared writer", 3)
 
 
class EventWriter:
    """Append ``TurnEvent``s published on an :class:`EventBus` to a rotating NDJSON log."""
 
    def __init__(
        self,
        base_dir: Path,
        *,
        max_file_bytes: int = 10 * 1024 * 1024,
        rotation_format: str = "{stem}.{seq:04d}.jsonl",
        stem: str = "events",
    ) -> None:
        self._base_dir = Path(base_dir)
        self._max_file_bytes = int(max_file_bytes)
        self._rotation_format = rotation_format
        self._stem = stem
        self._active_fd: Optional[IO[bytes]] = None
        self._active_path: Optional[Path] = None
        self._seq = 0
        self._subscription: Optional[Subscription] = None
        self._bus: Optional[EventBus] = None
        self._lock = threading.RLock()
 
    @property
    def active_path(self) -> Optional[Path]:
        """The file the next event will be written to (``None`` until attached)."""
        return self._active_path
 
    def _slot_path(self, seq: int) -> Path:
        return self._base_dir / self._rotation_format.format(stem=self._stem, seq=seq)
 
    def _discover_resume_seq(self) -> int:
        """Return the highest existing rotation seq in ``base_dir`` (ADR-4), else 0."""
        pattern = re.compile(rf"^{re.escape(self._stem)}\.(\d+)\.jsonl$")
        max_seq = 0
        for child in self._base_dir.iterdir():
            if not child.is_file():
                continue
            match = pattern.match(child.name)
            if match:
                max_seq = max(max_seq, int(match.group(1)))
        return max_seq
 
    def attach(self, bus: EventBus) -> Subscription:
        """Subscribe ``_on_event`` to ``bus`` as a wildcard consumer; open the active file.
 
        Creates ``base_dir`` if missing and resumes from the highest existing
        rotation slot. Returns the :class:`Subscription` handle so callers can
        ``detach`` symmetrically.
        """
        func = "attach"
        with self._lock:
            self._base_dir.mkdir(parents=True, exist_ok=True)
            self._seq = self._discover_resume_seq()
            self._active_path = self._slot_path(self._seq)
            self._active_fd = open(self._active_path, "ab", buffering=0)
            self._bus = bus
            self._subscription = bus.subscribe("*", self._on_event)
        _dbg(
            "OC_CHAT.MAIN",
            func,
            f"attached base_dir={self._base_dir} active={self._active_path.name}",
            3,
        )
        return self._subscription
 
    def open_unsubscribed(self) -> "PreparedEventWriter":
        """Open the active file WITHOUT subscribing a bus — the PREPARE half.
 
        :meth:`attach` opens the file and THEN subscribes; a transition needs
        those two halves split, because the fallible I/O (mkdir, resume scan,
        open) must happen in PREPARE while the bus mutation is the no-fail COMMIT
        step (plan 152 ADR-6 / Codex R08-02).
 
        Returns a :class:`PreparedEventWriter` carrying an OWNERSHIP RECEIPT that
        records precisely which directories and which slot file THIS attempt
        created versus which already existed. If the transition later aborts,
        :meth:`PreparedEventWriter.abort` closes the descriptor and removes ONLY
        this attempt's own artifacts, and only while they are still empty — a
        pre-existing event log is never touched.
 
        No bus interaction has happened when this returns, so an abort has
        nothing to unsubscribe.
        """
        func = "open_unsubscribed"
        with self._lock:
            created_dirs = _missing_ancestors(self._base_dir)
            self._base_dir.mkdir(parents=True, exist_ok=True)
            self._seq = self._discover_resume_seq()
            self._active_path = self._slot_path(self._seq)
            file_preexisted = self._active_path.exists()
            self._active_fd = open(self._active_path, "ab", buffering=0)
        receipt = EventWriterOwnershipReceipt(
            created_dirs=created_dirs,
            created_file=None if file_preexisted else self._active_path,
            file_preexisted=file_preexisted,
        )
        _dbg(
            "OC_CHAT.MAIN",
            func,
            f"opened base_dir={self._base_dir} active={self._active_path.name} "
            f"created_dirs={len(created_dirs)} file_preexisted={file_preexisted}",
            3,
        )
        return PreparedEventWriter(writer=self, receipt=receipt)
 
    def subscribe_only(self, bus: EventBus) -> Subscription:
        """Subscribe an ALREADY-OPENED writer to *bus* — the COMMIT half.
 
        Complements :meth:`open_unsubscribed`. Used by
        :func:`oc_chat.core.event_bootstrap.install_prepared_writer`, which
        performs the linearizable swap itself; this only records the resulting
        handle on the writer so :meth:`detach` stays symmetric.
        """
        with self._lock:
            self._bus = bus
            self._subscription = bus.subscribe("*", self._on_event)
            return self._subscription
 
    def adopt_subscription(self, bus: EventBus, subscription: Subscription) -> None:
        """Record a subscription created on this writer's behalf by the bus swap.
 
        :meth:`EventBus.replace_subscription` builds the handle atomically, so the
        writer cannot create it itself; this stores the result so :meth:`detach`
        can unsubscribe symmetrically. Pure assignment — cannot fail.
        """
        with self._lock:
            self._bus = bus
            self._subscription = subscription
 
    def detach(self) -> None:
        """Unsubscribe from the bus and close the active file (flush + fsync first)."""
        func = "detach"
        with self._lock:
            if self._subscription is not None and self._bus is not None:
                self._bus.unsubscribe(self._subscription)
            self._subscription = None
            self._bus = None
            self._close_active()
        _dbg("OC_CHAT.MAIN", func, "detached", 3)
 
    def close(self) -> None:
        """Detach and ensure the active file is flushed, fsynced, and closed.
 
        Equivalent to :meth:`detach`; kept as an explicit lifecycle name for
        callers that own the writer outside a bus subscription.
        """
        self.detach()
 
    def _close_active(self) -> None:
        """Flush + fsync + close the active file handle if open. Caller holds the lock."""
        func = "_close_active"
        if self._active_fd is None:
            return
        try:
            self._active_fd.flush()
            os.fsync(self._active_fd.fileno())
        except OSError as exc:  # best-effort on teardown
            _dbg(
                "OC_CHAT.MAIN",
                func,
                f"flush/fsync on close failed: {type(exc).__name__}: {exc}",
                2,
            )
        finally:
            self._active_fd.close()
            self._active_fd = None
 
    def _rotate(self) -> None:
        """Close the active file and open the next sequence slot. Caller holds the lock."""
        func = "_rotate"
        old_name = self._active_path.name if self._active_path is not None else "?"
        self._close_active()
        self._seq += 1
        self._active_path = self._slot_path(self._seq)
        self._active_fd = open(self._active_path, "ab", buffering=0)
        _dbg("OC_CHAT.MAIN", func, f"rotated {old_name} -> {self._active_path.name}", 3)
 
    def _on_event(self, event: TurnEvent) -> None:
        """Serialize ``event`` to one canonical-JSON line and append it under the file lock.
 
        After the write, rotates in-line if the active file has reached
        ``max_file_bytes`` (ADR-3). Serialization / write / fsync failures are
        logged at L2 and re-raised (the bus isolates them per M2.1 ADR-3).
        """
        func = "_on_event"
        with self._lock:
            if self._active_fd is None:
                _dbg("OC_CHAT.MAIN", func, "event received while detached; ignoring", 3)
                return
            try:
                line = canonical_json(event.model_dump(mode="json")).encode("utf-8") + b"\n"
            except Exception as exc:  # noqa: BLE001 - surface then re-raise
                _dbg(
                    "OC_CHAT.MAIN",
                    func,
                    f"serialization failed kind={event.kind!r}: {type(exc).__name__}: {exc}",
                    2,
                )
                raise
            try:
                with _FileLock(self._active_fd):
                    self._active_fd.write(line)
                    self._active_fd.flush()
                    os.fsync(self._active_fd.fileno())
            except OSError as exc:
                _dbg(
                    "OC_CHAT.MAIN",
                    func,
                    f"write/fsync failed path={self._active_path}: {type(exc).__name__}: {exc}",
                    2,
                )
                raise
            if os.fstat(self._active_fd.fileno()).st_size >= self._max_file_bytes:
                self._rotate()
 
 
__all__ = ["EventWriter"]