Source code for GUIBRUSHR.Retrieval.debug_log

"""
Shared JSONL debug logger for the Retrieval pipeline.

Used by both ``ExofastMCMC.exofast_demc`` (parent MCMC driver) and
``ModelCalculation.ModelData.parallel_chain`` (parallel workers) so that
high-level events (setup, convergence, finalization) and low-level events
(per-core proposal / accept / chain_end) land in a single chronologically
ordered file suitable for offline analysis.

Concurrency
-----------
Multiprocessing workers append independently. Under POSIX a single
``write()`` call whose buffer is smaller than ``PIPE_BUF`` (4096 B on
Linux) is atomic — so by keeping each JSON line under ~3800 bytes we
avoid interleaved writes without any lock.

Robustness
----------
``emit_event`` never raises: a broken logger must never kill the MCMC.
"""

import json
import os
import time
from pathlib import Path


_DEBUG_LOG_MAX_LINE_BYTES = 3800  # headroom below POSIX PIPE_BUF (4096)




def get_log_path(model_obj):
    """Return the JSONL debug file path for this retrieval, or None.

    Args:
        model_obj: any object exposing ``retrieval_data.path_results``.

    Returns:
        pathlib.Path or None.
    """
    try:
        base = model_obj.retrieval_data.path_results
    except Exception:
        return None
    if not base:
        return None
    return Path(base) / "debug" / "retrieval_debug.jsonl"





def init_log(path, header):
    """Create the parent dir, truncate the file, write a header line."""
    if path is None:
        return
    try:
        p = Path(path)
        p.parent.mkdir(parents=True, exist_ok=True)
        with p.open("w", encoding="utf-8") as f:
            header = dict(header)
            header.setdefault("ts", time.time())
            header.setdefault("pid", os.getpid())
            f.write(json.dumps(header, default=str, ensure_ascii=False) + "\n")
    except Exception:
        pass



def _serialize_record(record):
    """Render one event dict to a single JSON line respecting the PIPE_BUF cap."""
    record.setdefault("ts", time.time())
    record.setdefault("pid", os.getpid())
    line = json.dumps(record, default=str, ensure_ascii=False) + "\n"
    if len(line.encode("utf-8")) > _DEBUG_LOG_MAX_LINE_BYTES:
        safe = {}
        for k, v in record.items():
            try:
                length = len(v)
            except TypeError:
                length = 0
            if length > 8 and hasattr(v, "__iter__"):
                safe[k] = f"<len={length}>"
            else:
                safe[k] = v
        safe["_truncated"] = True
        line = json.dumps(safe, default=str, ensure_ascii=False) + "\n"
    return line




def emit_event(path, record):
    """Append one JSON line to ``path`` atomically (POSIX, <3800 B).

    Args:
        path: str/Path of the JSONL log file, or None to disable.
        record: dict to serialize. 'ts' and 'pid' auto-filled if missing.
    """
    if path is None:
        return
    try:
        line = _serialize_record(record)
        with open(str(path), "a", encoding="utf-8") as f:
            f.write(line)
    except Exception:
        pass





class BufferedLogger:
    """In-memory batch of JSON events, flushed on demand.

    Use this to decouple ``emit`` frequency from disk I/O: the parent MCMC
    driver accumulates events during a burnin block and flushes once at the
    distribution-save checkpoint (every 10 outer_steps). Workers use a
    short-lived instance and flush once before exiting their outer_step.

    Thread/process safety
    ---------------------
    Not shared across processes. Each process owns its own buffer. The single
    flush call is one ``write()`` per line under ``open(..., "a")``, so the
    POSIX <PIPE_BUF atomicity guarantee still holds line-by-line — workers
    flushing concurrently with the parent do not interleave within a line.
    """

    __slots__ = ("_lines",)



    def __init__(self):
        self._lines: list[str] = []




    def emit(self, record):
        """Append a record to the in-memory buffer. Never raises."""
        try:
            self._lines.append(_serialize_record(record))
        except Exception:
            pass


    def __len__(self):
        return len(self._lines)



    def flush(self, path):
        """Write all buffered lines to ``path`` (append mode) and clear.

        If ``path`` is None or the write fails, the buffer is cleared anyway
        to prevent unbounded memory growth on long runs.
        """
        if not self._lines:
            return
        lines = self._lines
        self._lines = []
        if path is None:
            return
        try:
            with open(str(path), "a", encoding="utf-8") as f:
                f.writelines(lines)
        except Exception:
            pass