event_bootstrap.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/event_bootstrap.py
 
"""
Per-process auto-attach of the M2.2 ``EventWriter`` to the default bus — M2.3.
 
Purpose
-------
The React loop (M2.3) publishes ``TurnEvent``s on ``default_bus()`` but does
not persist them. This module wires the M2.2 ``EventWriter`` to that bus once
per process at session bootstrap (plan 101 ADR-2): not at import time (would
touch the filesystem on import) and not lazily inside ``run_turn`` (would
couple the loop to disk state and complicate unit fixtures). The single
bootstrap call site is ``oc_chat/main.py``, immediately after the active
``ChatSession`` is bound.
 
Responsibilities
----------------
- :func:`attach_default_writer` — idempotently construct one ``EventWriter``
  rooted at the resolved base dir and attach it to ``default_bus()``.
- :func:`detach_default_writer` — close the active writer and clear the
  module singleton (test teardown + clean process exit).
- :func:`_resolve_base_dir` — honour ``OC_CHAT_EVENT_DIR`` else default to
  ``<cwd>/.oc_chat/events/<session-slug>``.
- :func:`_slug` — sanitise a session name into a path-safe slug.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — writer attach / detach lifecycle.
    L3 — base-dir resolution detail (env override vs default).
    L4 — module-load coverage marker.
 
Contracts
---------
- ``attach_default_writer`` is idempotent: repeated calls return the same
  ``EventWriter`` and never double-subscribe the bus.
- The writer is attached to the process-wide ``default_bus()`` singleton, so
  every event the React loop publishes is persisted.
- ``detach_default_writer`` is safe to call when no writer is attached (no-op).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.event_bus (default_bus — the process-wide event bus)
- oc_chat.core.event_writer (EventWriter — the rotating NDJSON sink)
- oc_chat.core.external_content_gate (lazy; canonical session id check for the event directory name)
- oc_chat.core.session (lazy; ChatSession type of the session whose events are written)
"""
from __future__ import annotations
 
import os
import re
from pathlib import Path
from typing import TYPE_CHECKING, Optional
 
from oc_diagnostics import _dbg
 
from oc_chat.core.event_bus import default_bus
from oc_chat.core.event_writer import EventWriter, PreparedEventWriter
 
if TYPE_CHECKING:
    from oc_chat.core.session import ChatSession
 
_dbg("OC_CHAT.MAIN", "event_bootstrap", "module", 4)
 
# Module-level singleton guarding re-attach attempts (ADR-2 idempotency).
_active_writer: Optional[EventWriter] = None
 
 
def _slug(name: str) -> str:
    """Return a path-safe slug for *name* (lowercase, non ``[a-z0-9_-]`` → ``-``)."""
    slug = re.sub(r"[^a-z0-9_-]+", "-", (name or "").lower()).strip("-")
    return slug or "session"
 
 
def _event_root() -> Path:
    """Return the ROOT under which per-session event dirs live (plan 152 D8).
 
    ``OC_CHAT_EVENT_DIR`` is a ROOT, not an exact per-process directory. It used
    to be returned verbatim, which meant every session in a process collided
    under one directory — incompatible with deriving the dir from the session
    identity. Callers append the session subdirectory themselves; the OLD direct
    root stays READABLE as a legacy candidate (see :func:`event_read_dirs`).
    """
    env_dir = os.environ.get("OC_CHAT_EVENT_DIR")
    if env_dir:
        return Path(env_dir)
    return Path.cwd() / ".oc_chat" / "events"
 
 
def _session_dir_key(session: "ChatSession") -> str:
    """The directory name for *session*: its canonical id, else the name slug.
 
    The canonical ``session_id`` is IMMUTABLE, so deriving the event dir from it
    makes rename, restart and same-name reuse all stable — a name slug does not
    (it splits one session's history across a rename and collides when two
    sessions share a name). The slug remains the fallback for doubles that carry
    no canonical id.
    """
    from oc_chat.core.external_content_gate import is_canonical_session_id
    sid = getattr(session, "session_id", None)
    if is_canonical_session_id(sid):
        return str(sid)
    return _slug(getattr(session, "name", "") or "")
 
 
def _resolve_base_dir(session: "ChatSession") -> Path:
    """Resolve the writer base dir: ``<event root>/<session_id>`` (plan 152 D8).
 
    The WRITE dir is derived from the immutable canonical ``session_id``, not the
    mutable name slug. ``OC_CHAT_EVENT_DIR`` selects the root; the session
    subdirectory is appended under it either way, so the override can no longer
    collide across sessions.
    """
    func = "_resolve_base_dir"
    base = _event_root() / _session_dir_key(session)
    _dbg("OC_CHAT.MAIN", func, f"base={base}", 3)
    return base
 
 
def _valid_alias(value: object) -> bool:
    """True iff *value* is EXACTLY a canonical :func:`_slug` output (R11-08).
 
    The persisted ``legacy_event_slugs`` collection is untrusted path input:
    ``load_session_metadata`` returns the RAW dict when model validation fails,
    so model-level validation alone is bypassable and the grammar must be
    re-checked AT USE. Requiring ``value == _slug(value)`` and non-empty rejects
    separators, ``.``/``..``, absolute / drive / UNC forms, and every
    non-canonical spelling in one predicate, because ``_slug`` maps all of them
    to something else.
    """
    if not isinstance(value, str) or not value:
        return False
    return value == _slug(value)
 
 
def _contained_alias_dir(root: Path, alias: str) -> Optional[Path]:
    """Resolve *alias* under *root* and prove containment AFTER resolution.
 
    Returns the resolved directory, or ``None`` when the alias is not a valid
    slug, does not exist, or resolves OUTSIDE the event root (the pinned
    symlink / reparse-point policy: comparison is on RESOLVED real paths, so a
    link escaping the root is rejected rather than followed).
    """
    func = "_contained_alias_dir"
    if not _valid_alias(alias):
        _dbg("OC_CHAT.MAIN", func, "rejected alias: not a canonical slug", 3)
        return None
    try:
        root_resolved = root.resolve(strict=False)
        candidate = (root / alias).resolve(strict=False)
        if not candidate.is_relative_to(root_resolved):
            _dbg("OC_CHAT.MAIN", func, "rejected alias: resolves outside the event root", 2)
            return None
        if not candidate.is_dir():
            return None
        return candidate
    except Exception as exc:
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: alias resolution failed: {type(exc).__name__}", 3)
        return None
 
 
def event_read_dirs(session: "ChatSession") -> tuple[Path, ...]:
    """Return the CANDIDATE event-log read roots for *session*, in MERGE ORDER.
 
    A directory is a candidate SOURCE, never an identity — the reader must still
    filter every line by payload ``session_id`` (real slug dirs are measurably
    multi-session). This function only says WHERE to look.
 
    Order is the pinned total order (plan 152 ADR-8): ALL legacy roots precede
    the ``session_id`` dir, because post-cutover writes go only to the id dir —
    a migration-epoch fact, not a timestamp guess. Across legacy roots the
    precedence is: the default name-slug dir, then the persisted
    ``legacy_event_slugs`` aliases in their recorded order, then the OLD direct
    ``OC_CHAT_EVENT_DIR`` root.
 
    Discovery is READ-ONLY: this never creates, mutates or dirties anything.
    Only existing directories are returned, de-duplicated by resolved path.
    """
    func = "event_read_dirs"
    root = _event_root()
    out: list[Path] = []
    seen: set[str] = set()
 
    def _add(path: Optional[Path]) -> None:
        if path is None:
            return
        try:
            if not path.is_dir():
                return
            key = str(path.resolve(strict=False)).casefold()   # Windows-safe identity
        except Exception:
            return
        if key in seen:
            return
        seen.add(key)
        out.append(path)
 
    id_key = _session_dir_key(session)
    name_slug = _slug(getattr(session, "name", "") or "")
 
    # 1. legacy: the pre-migration name-slug dir (skip when it IS the id dir).
    if name_slug != id_key:
        _add(root / name_slug)
    # 2. legacy: persisted rename aliases, re-validated + containment-proved.
    for alias in getattr(session, "legacy_event_slugs", None) or ():
        _add(_contained_alias_dir(root, alias))
    # 3. legacy: the OLD direct override root (events written before it became a root).
    if os.environ.get("OC_CHAT_EVENT_DIR"):
        _add(root)
    # 4. the current identity dir, LAST.
    _add(root / id_key)
 
    _dbg("OC_CHAT.MAIN", func, f"{len(out)} candidate root(s)", 3)
    return tuple(out)
 
 
def attach_default_writer(
    session: "ChatSession",
    *,
    base_dir: Optional[Path] = None,
) -> Optional[EventWriter]:
    """Attach a singleton ``EventWriter`` to ``default_bus()`` (idempotent).
 
    Returns the active writer. A second call returns the same instance without
    re-subscribing the bus, so the wildcard subscriber count stays at one.
    """
    func = "attach_default_writer"
    global _active_writer
    if _active_writer is not None:
        return _active_writer
    base = base_dir or _resolve_base_dir(session)
    writer = EventWriter(base)
    writer.attach(default_bus())
    _active_writer = writer
    _dbg("OC_CHAT.MAIN", func, f"attached writer at {base}", 2)
    return writer
 
 
def active_event_dir(session: "ChatSession") -> Path:
    """Return the session's event-log base dir (read-only; M2.6 consumers).
 
    Prefers the attached writer's actual base dir so ``/trace`` and ``/packet``
    read exactly where events are being written; falls back to
    :func:`_resolve_base_dir` when no writer is attached (e.g. tests).
    """
    if _active_writer is not None:
        return _active_writer._base_dir
    return _resolve_base_dir(session)
 
 
def install_prepared_writer(prepared: "PreparedEventWriter") -> EventWriter:
    """Swap the singleton writer for a PREPARE-opened one — COMMIT. No-fail.
 
    The bus mutation is a single :meth:`EventBus.replace_subscription` under ONE
    lock acquisition, so a ``publish`` racing this swap observes EXACTLY one
    wildcard writer — never zero, never two. All the fallible work (mkdir,
    resume scan, open) already happened in
    :meth:`EventWriter.open_unsubscribed`; what remains is a lock-protected list
    mutation and two assignments, none of which can raise.
 
    The OLD writer is closed immediately after the swap. There is no in-flight
    drain: production publishers are main-thread and between-turns quiescent
    (see the ``event_bus`` module Contracts note), and the transition's entry
    lease forecloses a pre-swap snapshot. Closing is best-effort — the old
    writer is already unsubscribed, so it cannot route later events either way.
 
    Linearization boundary: pre-swap events are in the OLD dir, post-swap events
    in the NEW dir.
    """
    func = "install_prepared_writer"
    global _active_writer
    old = _active_writer
    old_sub = getattr(old, "_subscription", None) if old is not None else None
    bus = default_bus()
    new_sub = bus.replace_subscription(old_sub, "*", prepared.writer._on_event)
    prepared.writer.adopt_subscription(bus, new_sub)
    _active_writer = prepared.writer
    if old is not None:
        try:
            old.close()
        except Exception as exc:
            _dbg("OC_CHAT.MAIN", func, f"EXTERNAL_ERROR: old writer close failed: {exc}", 2)
    _dbg("OC_CHAT.MAIN", func, f"installed writer at {prepared.writer._base_dir}", 2)
    return prepared.writer
 
 
def detach_default_writer() -> None:
    """Close the active writer and clear the singleton. No-op when unattached."""
    func = "detach_default_writer"
    global _active_writer
    if _active_writer is not None:
        _active_writer.close()
        _active_writer = None
        _dbg("OC_CHAT.MAIN", func, "detached writer", 2)
 
 
__all__ = ["attach_default_writer", "detach_default_writer", "active_event_dir"]