Bound JSONL growth with lossless process rotation; release 0.2.25

docs/README.md

@@ -430,4 +430,6 @@ pip install -e ".[dev]"
 | `examples/` | Usage examples | [Browse](./examples) |
 | `mkdocs.yml` | MkDocs configuration | — |
 
-<!-- code2docs:end -->
+<!-- code2docs:end -->
+
+- [Bounded JSONL writes and cooperative rotation](information/jsonl-rotation.md)

docs/information/jsonl-rotation.md

@@ -0,0 +1,124 @@
+---
+{
+  "schema": "wellmanifest.docs/document/v1",
+  "id": "jsonl-rotation",
+  "kind": "information",
+  "version": 1,
+  "title": "Bounded JSONL writes and cooperative rotation",
+  "status": "proposed",
+  "owner": "semcod/nfo",
+  "scope": "repository",
+  "created": "2026-11-05",
+  "updated": "2026-11-05",
+  "review_after": "2026-12-05",
+  "source_revision": "2d74ad212ef279aef19b7a3aa9a6a507b2ead5af",
+  "affected_repositories": [
+    "semcod/nfo"
+  ],
+  "evidence": [
+    "https://github.com/semcod/nfo/issues/10",
+    "repo://semcod/nfo/nfo/json_rotation.py",
+    "repo://semcod/nfo/tests/test_json_sink_rotation.py"
+  ]
+}
+---
+
+# Bounded JSONL writes and cooperative rotation
+
+<!-- docs:section purpose -->
+## Purpose
+
+Keep an explicitly configured active JSONL file within a byte budget while
+preserving historical bytes and oversized events in separate raw segments.
+This addresses the producer dependency of maskservice/c2004#738 (PLF-22243).
+
+<!-- docs:section scope -->
+## Scope
+
+`JSONSink` keeps its existing unbounded default. Bounded mode is opt-in and
+currently requires POSIX `flock`, directory descriptors and atomic same-device
+rename. Unsupported operating systems refuse bounded mode explicitly; the
+legacy sink remains available. This source change does not activate any service,
+change a runtime package or grant publication or maintenance authority.
+
+<!-- docs:section evidence -->
+## Evidence
+
+Tests exercise multiple spawned processes, unchanged legacy behavior,
+oversized UTF-8 events, partial existing tails, short writes, process death
+immediately after rollover, nonregular and linked paths, and refused I/O.
+The required protected matrix remains Python 3.11, 3.12 and 3.13. Passing local
+tests does not establish independently approved merge or deployment.
+
+<!-- docs:section content -->
+## Configuration and protocol
+
+Use `JSONSink(path, max_bytes=50000000)`, or set `NFO_JSONL_MAX_BYTES=50000000`
+in the intended service environment. The constructor argument overrides the
+environment. Zero disables bounding explicitly; unset environment also disables
+it. Negative, boolean or noninteger constructor values and invalid environment
+values are refused before writing. The budget counts encoded UTF-8 bytes,
+including the final newline, rather than characters.
+
+Every writer to a path must use bounded mode. Before enabling it, stop legacy
+writers and activate the cooperating implementation together. Do not combine
+it with an offline rotator or an external tool that ignores the lock. Do not
+remove, rename, replace or hardlink the persistent `.<filename>.nfo.lock`.
+
+An instance thread lock protects serialization inside its process. A stable
+per-path `flock` then serializes processes before they open the active inode.
+The active file is opened anew for each event, so a waiting writer follows the
+new inode after rollover. Symlinked path components, symlinked files, special
+files and multiply linked active/lock files are refused. Create the log's parent
+directory before use. New active files and lock files use mode 0600.
+
+Readers outside the lock may briefly observe a missing active pathname between
+rollover and recreation; treat that as an empty current file and retry, while
+historical bytes remain in the segment. Writers following this protocol never
+open or append the active file outside the shared lock.
+
+Before a complete event would exceed the budget, the current nonempty file is
+fsynced and moved atomically into an exclusively created private directory under
+`archive/<filename>.nfo-<time>-<uuid>/events.jsonl`. Historical bytes are not
+parsed, truncated, rewritten or deleted. An incomplete existing tail also causes
+rollover before a new event, preserving the opaque tail without joining the new
+JSON record to it. Each private segment directory uses mode 0700.
+
+An event larger than the entire budget is preserved intact in its own segment;
+it is not dropped or split and the active file remains empty. A partially failed
+oversized write remains `.events.incomplete` for inspection and is not named as
+a completed segment. Short writes are retried. The downstream delegate runs only
+after the local write/segment has completed successfully.
+
+Active and segment file data are fsynced, as are the changed directories.
+Segments remain raw rather than being compressed on the append path. This keeps
+rollover short; later backup/compression/retention is a separate operation.
+There is no archive count limit, automatic pruning or disk-space quota. Segment
+names are identifiers, not a trusted global ordering index across clock changes;
+bytes and record order within each segment remain intact.
+
+<!-- docs:section limitations -->
+## Failure and compatibility boundaries
+
+All writers must cooperate and use a consistent budget. A thread lock alone,
+a legacy writer or another program ignoring the sidecar can invalidate the
+bound. A successful write by cooperating processes leaves the active file at
+or below the configured budget. Archives can exceed that budget and continue
+to consume disk space; the budget is for the active file only.
+
+A process dying after atomic rollover leaves its old bytes in a segment and
+releases its OS lock; a subsequent cooperating writer can create the new active
+file. An I/O or fsync error is a failed operation. The active path, a raw segment
+or an incomplete artifact may remain; inspect before retrying. No exactly-once
+promise is made after uncertain durability or partial I/O. Never delete recovery
+artifacts automatically, and do not treat fixture tests as a power-loss test.
+
+<!-- docs:section next_actions -->
+## Publication and activation
+
+Publish through the repository's existing protected Python matrix and independent
+Validator. Activate an immutable approved package in the intended service only,
+with all previous writers stopped and recoverable state available. Check the
+actual loaded package, scoped environment, active-file size and service health
+after restart. Existing archives must remain preserved. Other services keep
+the legacy default unless explicitly configured.

nfo/__init__.py

@@ -128,7 +128,7 @@ def __getattr__(name: str):
         return FastAPIMiddleware
     raise AttributeError(f"module 'nfo' has no attribute {name!r}")
 
-__version__ = "0.2.24"
+__version__ = "0.2.25"
 
 __all__ = [
     "log_call",

nfo/json_rotation.py

@@ -0,0 +1,129 @@
+"""Opt-in POSIX JSONL bounds with a stable lock and lossless raw segments."""
+from contextlib import contextmanager
+import os
+from pathlib import Path
+import stat
+import time
+from uuid import uuid4
+
+
+@contextmanager
+def _directory(path, create=False):
+    path = Path(path)
+    fd = os.open(path.anchor, os.O_RDONLY | os.O_DIRECTORY)
+    try:
+        for part in path.parts[1:]:
+            if create:
+                try:
+                    os.mkdir(part, mode=0o700, dir_fd=fd)
+                except FileExistsError:
+                    pass
+            child = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd)
+            os.close(fd)
+            fd = child
+        yield fd
+    finally:
+        os.close(fd)
+
+
+@contextmanager
+def _file(parent, name, flags):
+    fd = os.open(name, flags | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600, dir_fd=parent)
+    try:
+        info = os.fstat(fd)
+        if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1:
+            raise ValueError("bounded JSONL requires single-link regular files")
+        yield fd
+    finally:
+        os.close(fd)
+
+
+def _write_all(fd, data):
+    remaining = memoryview(data)
+    while remaining:
+        count = os.write(fd, remaining)
+        if count <= 0:
+            raise OSError("JSONL write made no progress")
+        remaining = remaining[count:]
+    os.fsync(fd)
+
+
+class BoundedJSONLWriter:
+    """All writers to a path must use this protocol; lock files stay in place.
+
+    Segments are not pruned or compressed on the append path. A segment directory
+    can survive a failed operation; its bytes must never be blindly removed.
+    """
+
+    def __init__(self, file_path, max_bytes):
+        if os.name != "posix":
+            raise NotImplementedError("bounded JSONL currently requires POSIX file locking")
+        import fcntl
+        self._fcntl = fcntl
+        self.path = Path(os.path.abspath(file_path))
+        self.max_bytes = max_bytes
+
+    @contextmanager
+    def _segment(self):
+        archive = self.path.parent / "archive"
+        with _directory(archive, create=True) as parent:
+            name = f"{self.path.name}.nfo-{time.time_ns()}-{uuid4().hex}"
+            os.mkdir(name, mode=0o700, dir_fd=parent)
+            folder = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent)
+            try:
+                yield folder
+                os.fsync(folder)
+                os.fsync(parent)
+            finally:
+                os.close(folder)
+
+    def _roll(self, parent, active):
+        # No content copying, truncation or history rewriting. The exclusively
+        # created private directory makes the destination unique and empty.
+        before = os.fstat(active)
+        current = os.stat(self.path.name, dir_fd=parent, follow_symlinks=False)
+        if (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) != (
+                current.st_dev, current.st_ino, current.st_size, current.st_mtime_ns):
+            raise OSError("JSONL path changed outside the cooperative lock")
+        os.fsync(active)
+        with self._segment() as folder:
+            os.rename(self.path.name, "events.jsonl", src_dir_fd=parent, dst_dir_fd=folder)
+        os.fsync(parent)
+
+    def write(self, payload):
+        with _directory(self.path.parent) as parent:
+            lock_name = "." + self.path.name + ".nfo.lock"
+            with _file(parent, lock_name, os.O_CREAT | os.O_RDWR) as lock:
+                self._fcntl.flock(lock, self._fcntl.LOCK_EX)
+                info = os.fstat(lock)
+                named = os.stat(lock_name, dir_fd=parent, follow_symlinks=False)
+                if (info.st_dev, info.st_ino) != (named.st_dev, named.st_ino):
+                    raise OSError("JSONL lock path changed")
+                self._append(parent, payload)
+
+    def _append(self, parent, payload):
+        flags = os.O_CREAT | os.O_RDWR | os.O_APPEND
+        with _file(parent, self.path.name, flags) as active:
+            size = os.fstat(active).st_size
+            partial = size > 0 and os.pread(active, 1, size - 1) != b"\n"
+            rollover = size > 0 and (partial or size + len(payload) > self.max_bytes)
+            if rollover:
+                self._roll(parent, active)
+            elif len(payload) <= self.max_bytes:
+                _write_all(active, payload)
+                os.fsync(parent)
+                return
+        # The old descriptor is closed before opening the fresh active inode.
+        with _file(parent, self.path.name, flags) as active:
+            if os.fstat(active).st_size:
+                raise OSError("fresh JSONL file changed outside the cooperative lock")
+            if len(payload) > self.max_bytes:
+                with self._segment() as folder:
+                    with _file(folder, ".events.incomplete", os.O_CREAT | os.O_EXCL | os.O_WRONLY) as out:
+                        _write_all(out, payload)
+                    os.rename(".events.incomplete", "events.jsonl",
+                              src_dir_fd=folder, dst_dir_fd=folder)
+                os.fsync(active)
+            else:
+                _write_all(active, payload)
+            os.fsync(parent)

nfo/json_sink.py

@@ -33,6 +33,9 @@ class JSONSink(Sink):
         file_path: Path to the JSON Lines file (default: ``logs.jsonl``).
         pretty: If True, indent JSON for readability (not recommended for production).
         delegate: Optional downstream sink to forward entries to.
+        max_bytes: Optional POSIX active-file byte budget. None reads
+            NFO_JSONL_MAX_BYTES (unset means zero); explicit zero keeps legacy
+            unbounded writes. All writers to a bounded path must cooperate.
     """
 
     def __init__(
@@ -42,12 +45,25 @@ def __init__(
         pretty: bool = False,
         compact: bool = False,
         delegate: Optional[Sink] = None,
+        max_bytes: Optional[int] = None,
     ) -> None:
         self.file_path = str(file_path)
         self.pretty = pretty
         self.compact = compact
         self.delegate = delegate
         self._lock = threading.Lock()
+        if max_bytes is None:
+            raw = os.environ.get("NFO_JSONL_MAX_BYTES", "0")
+            try:
+                max_bytes = int(raw)
+            except (ValueError, TypeError) as exc:
+                raise ValueError("NFO_JSONL_MAX_BYTES must be a nonnegative integer") from exc
+        if type(max_bytes) is not int or max_bytes < 0:
+            raise ValueError("max_bytes must be a nonnegative integer")
+        self._bounded = None
+        if max_bytes:
+            from nfo.json_rotation import BoundedJSONLWriter
+            self._bounded = BoundedJSONLWriter(self.file_path, max_bytes)
 
     def write(self, entry: LogEntry) -> None:
         d = entry.as_compact() if self.compact else entry.as_dict()
@@ -60,8 +76,11 @@ def write(self, entry: LogEntry) -> None:
         line = json.dumps(d, ensure_ascii=False, default=str, indent=indent)
 
         with self._lock:
-            with open(self.file_path, "a", encoding="utf-8") as f:
-                f.write(line + "\n")
+            if self._bounded is not None:
+                self._bounded.write((line + "\n").encode("utf-8"))
+            else:
+                with open(self.file_path, "a", encoding="utf-8") as f:
+                    f.write(line + "\n")
 
         if self.delegate:
             self.delegate.write(entry)

project/ticket-005/README.md

@@ -0,0 +1,16 @@
+# ticket-005: Bounded, lossless JSONL producer
+
+- Status: IN_PROGRESS
+- Workflow state: EDIT
+
+Planfile PLF-005; [issue #10](https://github.com/semcod/nfo/issues/10); dependency of maskservice/c2004#738.
+
+SESSION_EXECUTION_AUTHORIZATION: user continued the accepted logger-bound task.
+
+AC-01: Explicit constructor/environment opt-in, legacy default unchanged.
+AC-02: Persistent cross-process path lock before opening active file.
+AC-03: Complete raw segments and oversized records preserved; no archive pruning.
+AC-04: Concurrent writers, process death, opaque tails, symlinks and I/O failure checks.
+AC-05: Exact protected Python3.11/3.12/3.13 checks and independent Validator publication.
+
+Local qualification: 54 tests pass; pinned Docs completion passes. Frozen oldsource2d74 fails20 regression checks (0 setup errors). Release0.2.25. Protectedmatrix andindependentValidator remain pending; notDONE.

project/ticket-005/intent.json

@@ -0,0 +1,25 @@
+{
+  "ticket": "ticket-005",
+  "planfile_id": "PLF-005",
+  "summary": "Bound JSONL writes with cooperative process locking and lossless segments",
+  "allowedPaths": [
+    "nfo/json_sink.py",
+    "nfo/json_rotation.py",
+    "tests/test_json_sink_rotation.py",
+    "docs/information/jsonl-rotation.md",
+    "docs/README.md",
+    "project/ticket-005/README.md",
+    "project/ticket-005/intent.json",
+    "pyproject.toml",
+    "nfo/__init__.py"
+  ],
+  "forbiddenPaths": [
+    ".env",
+    ".github/**",
+    ".governance/**",
+    "errors/**"
+  ],
+  "authorization": "SESSION_EXECUTION_AUTHORIZATION: continuation of maskservice/c2004 PLF-22243/#738; dependency tracked by semcod/nfo#10",
+  "outcome": "Optional JSONSink byte bound survives concurrent writes without truncation; legacy default unchanged; protected3.11/3.12/3.13 matrix and independent Validator required.",
+  "release": "Bump authoritativepyprojectandimportversion0.2.24to0.2.25 under existinguserpublication+bump instruction."
+}

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "nfo"
-version = "0.2.24"
+version = "0.2.25"
 description = "Automatic function logging system with decorators, supporting multiple output sinks (SQLite, CSV, Markdown, Prometheus) and LLM-powered analysis for DevOps observability."
 readme = "README.md"
 license = {text = "Apache-2.0"}