feat(m1): add append-only jsonl writer with folding reader

scripts/jobbsok_lib/jsonl.py

@@ -0,0 +1,207 @@
+"""The append-only store for logg.jsonl, beslutninger.jsonl and varianter.jsonl.
+
+Append-only is the whole design, and it has a consequence people usually
+discover too late: the schema is irreversible. There is no migration that can
+go back and add a field to lines already written, so `type` and `skjema` are
+on the very first line this module ever writes (risk H10). A reader that meets
+a line without them is meeting a file this module did not produce.
+
+Two writers, one file (risk H4). The CLI and the host MCP server can both be
+appending while the operator watches, so every record goes out as a single
+``os.write`` on a descriptor opened ``O_APPEND``. That is what makes the
+append indivisible; building the line first and writing it in one call is not
+a style choice here, it is the guarantee. Anything that split the record into
+two writes -- a buffered file object, a separate newline write -- would let a
+second writer land between them.
+
+Torn writes versus corruption (risk M5r). A process killed mid-append leaves a
+partial last line with no newline after it. That is recoverable and expected,
+so :func:`read_lines` skips it. A malformed line anywhere else is corruption,
+not a torn write, and it raises with its line number rather than being
+silently dropped -- the difference between "the last append did not finish"
+and "a record in the middle of your decision log is gone" is the difference
+this module refuses to blur.
+
+Timestamps carry an explicit UTC offset, always. Norwegian local time repeats
+02:30 on the last Sunday of October, so two records an hour apart share a wall
+clock and are told apart only by the offset. A naive timestamp is rejected on
+write, because by read time there is no way to recover which hour was meant.
+"""
+
+import datetime
+import json
+import os
+
+from . import paths
+
+#: The closed set of record kinds. `beslutning` is a sourcing decision,
+#: `korrigering` restates one that was wrong, `utfall` records how a case
+#: ended. Anything else is a caller bug, not a new kind.
+TYPES = ("beslutning", "korrigering", "utfall")
+
+#: Schema version stamped on every line. Bumping this is a fork of the format,
+#: never an in-place migration -- the file behind it cannot be rewritten.
+SKJEMA = 1
+
+#: Keys whose values are timestamps, wherever they appear. `ts` is the case
+#: log (build-brief 5.3), `dato` the decision log (5.4).
+TIMESTAMP_KEYS = ("ts", "dato")
+
+#: Fields belonging to the correction record itself, not to the decision it
+#: corrects. Folding must not smear these onto the target.
+_CORRECTION_OWN_FIELDS = ("type", "id", "korrigerer", "skjema")
+
+
+class JsonlError(Exception):
+    """A refusal from this module, with the line number when there is one."""
+
+    def __init__(self, message, line=None):
+        self.line = line
+        super().__init__("line %d: %s" % (line, message) if line else message)
+
+
+def append_line(path, obj, root=None):
+    """Append ``obj`` as exactly one line, in a single ``O_APPEND`` write.
+
+    Validation happens before the file is opened, so a rejected record leaves
+    no trace at all -- not an empty file, not a partial line.
+
+    ``root``, when given, resolves ``path`` under it with
+    :func:`paths.safe_join`, so a caller-supplied path that escapes the
+    workspace is refused here rather than in whichever script remembered to
+    check.
+    """
+    record = _validated(obj)
+    target = paths.safe_join(root, path) if root is not None else path
+    payload = (json.dumps(record, ensure_ascii=False) + "\n").encode("utf-8")
+
+    handle = os.open(target, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600)
+    try:
+        written = os.write(handle, payload)
+    finally:
+        os.close(handle)
+    if written != len(payload):
+        # Retrying the remainder would append a fragment behind whatever
+        # another writer put there in the meantime. Fail instead.
+        raise JsonlError(
+            "short write: %d of %d bytes reached %r" % (written, len(payload), target)
+        )
+    return record
+
+
+def read_lines(path):
+    """Read every record, skipping only a torn trailing line.
+
+    A missing file reads as no records -- an append-only log that has never
+    been appended to is empty, not broken.
+    """
+    if not os.path.exists(path):
+        return []
+    with open(path, "r", encoding="utf-8") as handle:
+        data = handle.read()
+    if data == "":
+        return []
+
+    lines = data.split("\n")
+    # A file ending in a newline leaves an empty final element; a file that
+    # does not was cut off mid-append, and only THAT last line may be skipped.
+    if lines and lines[-1] == "":
+        lines.pop()
+        torn_index = None
+    else:
+        torn_index = len(lines)
+
+    records = []
+    for index, line in enumerate(lines, start=1):
+        if line.strip() == "":
+            continue
+        try:
+            record = json.loads(line)
+        except ValueError as exc:
+            if index == torn_index:
+                continue
+            raise JsonlError("not valid JSON: %s" % exc, line=index)
+        if not isinstance(record, dict):
+            if index == torn_index:
+                continue
+            raise JsonlError("expected an object, got %s" % type(record).__name__, line=index)
+        records.append(record)
+    return records
+
+
+def fold(lines):
+    """Return the decision stream with corrections applied and outcomes removed.
+
+    A `korrigering` overwrites the fields it names on the `beslutning` it
+    points at, and contributes no row of its own -- the log keeps both, the
+    stream shows the corrected decision. An `utfall` is not a decision and is
+    excluded; build-brief 5.7 reads outcomes on its own terms.
+
+    The input is never mutated: the caller's records came off disk and stay as
+    they were read.
+    """
+    stream = []
+    by_id = {}
+    for record in lines:
+        kind = record.get("type")
+        if kind == "beslutning":
+            copy = dict(record)
+            stream.append(copy)
+            if "id" in copy:
+                by_id[copy["id"]] = copy
+        elif kind == "korrigering":
+            target_id = record.get("korrigerer")
+            if target_id not in by_id:
+                raise JsonlError(
+                    "correction %r names %r, which is not a decision above it"
+                    % (record.get("id"), target_id)
+                )
+            by_id[target_id].update(
+                {k: v for k, v in record.items() if k not in _CORRECTION_OWN_FIELDS}
+            )
+        elif kind == "utfall":
+            continue
+        else:
+            raise JsonlError("unknown record type %r" % (kind,))
+    return stream
+
+
+def instant(record):
+    """The record's timestamp as an aware :class:`datetime.datetime`."""
+    for key in TIMESTAMP_KEYS:
+        if key in record:
+            return _aware(record[key], key)
+    raise JsonlError("record has no timestamp; expected one of %r" % (TIMESTAMP_KEYS,))
+
+
+def by_time(lines):
+    """Sort records by the instant they name, not by the string that spells it."""
+    return sorted(lines, key=instant)
+
+
+def _validated(obj):
+    if not isinstance(obj, dict):
+        raise JsonlError("a record must be an object, got %s" % type(obj).__name__)
+    kind = obj.get("type")
+    if kind not in TYPES:
+        raise JsonlError("type %r is not one of %r" % (kind, TYPES))
+
+    record = dict(obj)
+    record.setdefault("skjema", SKJEMA)
+    for key in TIMESTAMP_KEYS:
+        if key in record:
+            _aware(record[key], key)
+    return record
+
+
+def _aware(value, key):
+    try:
+        moment = datetime.datetime.fromisoformat(value)
+    except (TypeError, ValueError):
+        raise JsonlError("%s=%r is not an ISO-8601 timestamp" % (key, value))
+    if moment.tzinfo is None or moment.utcoffset() is None:
+        raise JsonlError(
+            "%s=%r has no UTC offset; a naive timestamp cannot be ordered "
+            "across the October hour that happens twice" % (key, value)
+        )
+    return moment

tests/test_jsonl.py

@@ -0,0 +1,133 @@
+"""The append-only store behind logg.jsonl, beslutninger.jsonl, varianter.jsonl
+(plan Step 5).
+
+Three risks meet in this one module. Risk H4: the CLI and the host MCP server
+can both be writing while the operator watches, so an append that is not a
+single O_APPEND write can interleave two records into one unreadable line.
+Risk H10: the store is append-only, so its schema is irreversible -- `type`
+and `skjema` have to be on the very first line ever written, because there is
+no migration that can add them later. Risk M5r: a torn trailing line from a
+killed process is recoverable and must not take the whole log with it, while a
+malformed line in the middle is corruption and must stop the run loudly.
+
+The DST test is the one that looks like trivia and is not. Norwegian local
+time repeats 02:30 on the last Sunday of October; two records written an hour
+apart carry the same wall clock and differ only in the offset. Sorting on the
+string would call them simultaneous.
+
+Style note: this file follows tests/test_frontmatter.py.
+"""
+
+import json
+import os
+import threading
+
+import pytest
+
+from jobbsok_lib import jsonl
+
+
+def entry(**overrides):
+    record = {
+        "type": "beslutning",
+        "id": "b1",
+        "dato": "2026-09-15T09:00:00+02:00",
+        "beslutning": "ja",
+    }
+    record.update(overrides)
+    return record
+
+
+def test_append_writes_one_line_and_leaves_prior_bytes_untouched(tmp_path):
+    path = str(tmp_path / "beslutninger.jsonl")
+    jsonl.append_line(path, entry(id="b1"))
+    before = open(path, "rb").read()
+
+    jsonl.append_line(path, entry(id="b2"))
+    after = open(path, "rb").read()
+
+    # Not "the file still parses" -- the earlier bytes must be the same bytes.
+    assert after.startswith(before)
+    assert after[len(before):].count(b"\n") == 1
+    assert [rec["id"] for rec in jsonl.read_lines(path)] == ["b1", "b2"]
+
+
+def test_two_concurrent_writers_produce_two_intact_lines(tmp_path):
+    path = str(tmp_path / "logg.jsonl")
+    start = threading.Barrier(2)
+    count = 25
+
+    def writer(tag):
+        start.wait()
+        for index in range(count):
+            jsonl.append_line(path, entry(id="%s-%d" % (tag, index), notat="x" * 180))
+
+    threads = [threading.Thread(target=writer, args=(tag,)) for tag in ("a", "b")]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    records = jsonl.read_lines(path)
+    assert len(records) == 2 * count
+    assert len({rec["id"] for rec in records}) == 2 * count
+
+
+def test_a_torn_trailing_line_is_skipped_not_fatal(tmp_path):
+    path = str(tmp_path / "logg.jsonl")
+    jsonl.append_line(path, entry(id="b1"))
+    with open(path, "a", encoding="utf-8") as handle:
+        handle.write('{"type": "beslutning", "id": "b2", "dat')  # killed mid-write
+
+    records = jsonl.read_lines(path)
+    assert [rec["id"] for rec in records] == ["b1"]
+
+
+def test_a_malformed_interior_line_raises_with_its_line_number(tmp_path):
+    path = str(tmp_path / "logg.jsonl")
+    jsonl.append_line(path, entry(id="b1"))
+    with open(path, "a", encoding="utf-8") as handle:
+        handle.write("{ not json at all }\n")
+    jsonl.append_line(path, entry(id="b3"))
+
+    with pytest.raises(jsonl.JsonlError) as caught:
+        jsonl.read_lines(path)
+    assert caught.value.line == 2
+
+
+def test_fold_applies_a_correction_and_drops_an_outcome(tmp_path):
+    lines = [
+        entry(id="b1", beslutning="ja", arsak=["lonn"]),
+        entry(id="b2", beslutning="ja"),
+        entry(type="korrigering", id="k1", korrigerer="b1", beslutning="nei"),
+        entry(type="utfall", id="u1", korrigerer="b2", stadium="screening"),
+    ]
+    folded = jsonl.fold(lines)
+
+    assert [rec["id"] for rec in folded] == ["b1", "b2"]
+    # The correction overwrote the decision it names, and did not invent a row.
+    assert folded[0]["beslutning"] == "nei"
+    assert folded[0]["arsak"] == ["lonn"]
+    # The outcome is not part of the decision stream; §5.7 mines it separately.
+    assert all(rec["type"] == "beslutning" for rec in folded)
+
+
+def test_a_naive_timestamp_is_rejected_on_write(tmp_path):
+    path = str(tmp_path / "logg.jsonl")
+    with pytest.raises(jsonl.JsonlError):
+        jsonl.append_line(path, entry(dato="2026-09-15T09:00:00"))
+    # And nothing was written: a rejected record must not land half-way.
+    assert not os.path.exists(path) or open(path).read() == ""
+
+
+def test_timestamps_across_the_dst_boundary_sort_by_instant_not_by_string():
+    before_shift = entry(id="before", dato="2026-10-25T02:30:00+02:00")
+    after_shift = entry(id="after", dato="2026-10-25T02:30:00+01:00")
+
+    # Same wall clock, an hour apart. String order is alphabetical on the
+    # offset and happens to be wrong here.
+    assert sorted([after_shift["dato"], before_shift["dato"]])[0] == after_shift["dato"]
+    assert [rec["id"] for rec in jsonl.by_time([after_shift, before_shift])] == [
+        "before",
+        "after",
+    ]