From d4341fbc7a08c7b191c93ff543e86040a1b5a1fd Mon Sep 17 00:00:00 2001
From: Matt <matthew.field@ukaea.uk>
Date: Fri, 30 Oct 2026 13:53:07 +0000
Subject: [PATCH 1/8] initial commit of line numbering

---
 docs/custom_parsers.md      |  14 +++++
 docs/tailing.md             |  27 +++++++++
 multiparser/parsing/tail.py |  99 +++++++++++++++-----------------
 multiparser/thread.py       |  23 +++++++-
 multiparser/typing.py       |   2 +-
 tests/test_data_parsing.py  | 111 +++++++++++++++++++++++++++++++++++-
 6 files changed, 217 insertions(+), 59 deletions(-)

diff --git a/docs/custom_parsers.md b/docs/custom_parsers.md
index 0a0073d..aa6bede 100644
--- a/docs/custom_parsers.md
+++ b/docs/custom_parsers.md
@@ -78,6 +78,20 @@ with multiparser.FileMonitor(timeout=10) as monitor:
 In the case where you would like your parser function to accept additional keyword arguments you can add these
 to the definition and pass them to tracking using the `parser_kwargs` argument.
 
+### Line positions
+
+Tail parsers receive `__line_numbers` through their keyword arguments: a list of
+1-based physical file line numbers corresponding to the lines in `file_content`.
+Lines excluded by `skip_lines_w_pattern` have already been removed, but their
+positions still count, so the list may contain gaps.
+
+If a custom parser returns a list of row dictionaries, it can include
+`"__line_numbers": [...]` in its returned metadata to associate a position with
+each output row. The list must align with the **returned rows**, not necessarily
+all input lines. Each callback then receives its row's `metadata["line_number"]`.
+Existing custom parsers can ignore this feature and retain their current return
+format. Internal read-state metadata is maintained by Multiparser.
+
 ## Class methods as parsers
 
 When custom parsers are decorated any positional and keyword arguments are propagated through, i.e.:
diff --git a/docs/tailing.md b/docs/tailing.md
index 08c001b..a7499d8 100644
--- a/docs/tailing.md
+++ b/docs/tailing.md
@@ -77,6 +77,33 @@ Default: `None`
 
 Optional callback exclusively for this file set, this overwrites the global callback (if specified) within the `FileMonitor` definition (see [here](./file_monitor.md#per-thread-callback)).
 
+The built-in tail parsers (regular-expression logs, CSV, and other delimited files)
+include `metadata["line_number"]` in each callback. This is the **1-based physical
+file line**, counting headers, empty lines, and lines excluded by
+`skip_lines_w_pattern`. Numbering continues across appended batches and is
+independent for each file. As with tailing generally, this assumes append-only
+files with complete, newline-terminated records; CSV records spanning multiple
+physical lines are not supported.
+
+```python
+from multiparser.parsing.tail import record_csv
+
+
+def on_row(data, metadata):
+    print(metadata["file_name"], metadata["line_number"], data)
+
+
+monitor.tail(
+    path_glob_exprs="*.csv",
+    parser_func=record_csv,
+    callback=on_row,
+)
+```
+
+Custom block parsers are not assigned line numbers automatically, since an output
+row may represent several input lines. See [custom parsers](./custom_parsers.md#log-parsers)
+for explicit row-position support.
+
 
 ### `parser_func`
 `#!python Callable[[str, *_], tuple[dict[str, Any], dict[str, Any]]] | None`
diff --git a/multiparser/parsing/tail.py b/multiparser/parsing/tail.py
index 66606a5..e6fe168 100644
--- a/multiparser/parsing/tail.py
+++ b/multiparser/parsing/tail.py
@@ -259,17 +259,24 @@ def record_with_delimiter(
     """
     # The delimiter parser assumes each line is a new data entry so
     # revert back to list of lines here
-    _file_lines: list[str] = [i for i in file_content.split("\n") if i]
+    _file_lines = file_content.split("\n")
+    _line_numbers = kwargs.pop("__line_numbers", list(range(1, len(_file_lines) + 1)))
 
-    _parsed_data: tuple[dict[str, typing.Any], list[dict[str, typing.Any]]] = {}, []
+    _parsed_data: tuple[dict[str, typing.Any], list[dict[str, typing.Any]]] = (
+        {"__line_numbers": []},
+        [],
+    )
 
     if headers:
         _parsed_data[0]["headers"] = headers
 
-    if not _file_lines:
+    if not any(_file_lines):
         return {}, []
 
-    for file_line in _file_lines:
+    for line_number, file_line in zip(_line_numbers, _file_lines):
+        if not file_line:
+            continue
+        _parsed_data[0]["__line_numbers"].append(line_number)
         _parsed_line: TimeStampedData = _record_any_delimited(  # type: ignore[call-arg]
             file_content=file_line,
             delimiter=delimiter,
@@ -330,49 +337,14 @@ def record_csv(
         * metadata outlining properties such as modified time etc.
         * actual recorded data from the file.
     """
-    # The delimiter parser assumes each line is a new data entry so
-    # revert back to list of lines here
-    _file_lines: list[str] = [i for i in file_content.split("\n") if i]
-
-    _parsed_data: tuple[dict[str, typing.Any], list[dict[str, typing.Any]]] = {}, []
-
-    if headers:
-        _parsed_data[0]["headers"] = headers
-
-    if not _file_lines:
-        return {}, []
-
-    for file_line in _file_lines:
-        _parsed_line: TimeStampedData = _record_any_delimited(  # type: ignore[call-arg]
-            file_content=file_line,
-            delimiter=",",
-            tracked_values=tracked_values,
-            convert=convert,
-            **(_parsed_data[0] | kwargs),
-        )
-
-        if not isinstance(_parsed_line[1], dict):
-            raise AssertionError(
-                "Expected parsed statement to return a dictionary "
-                f"of recorded data but got {_parsed_line}"
-            )
-
-        # Make sure each line does not erase the previous metadata collected at the start
-        # of processing the block, e.g. if headers are set. May have further use in future
-        # if other info is extractable but not necessarily present in the first line
-        for key, value in _parsed_line[0].items():
-            if key in ("headers",) and not _parsed_data[0].get(key):
-                _parsed_data[0]["headers"] = value
-            else:
-                _parsed_data[0][key] = value
-        _parsed_data[1].append(_parsed_line[1])
-
-    # Headers must be read when the file is first created else any values after read
-    # will not align with these headings
-    if not _parsed_data[0].get("headers"):
-        raise AssertionError("Failed to retrieve file header during initial read")
-
-    return _parsed_data
+    return record_with_delimiter(
+        file_content=file_content,
+        delimiter=",",
+        headers=headers,
+        tracked_values=tracked_values,
+        convert=convert,
+        **kwargs,
+    )
 
 
 def _extract_label_value_pair(
@@ -526,6 +498,7 @@ def record_log(
     ignore_lines: list[re.Pattern[str]] | None = None,
     parser_func: ParserFunction | None = None,
     __read_bytes: int | None = None,
+    __lines_read: int = 0,
     **parser_kwargs,
 ) -> TimeStampedData:
     """Record lines within a log type file.
@@ -549,8 +522,13 @@ def record_log(
         specify an alternative tail parsing function
     __read_bytes : int | None, optional
         internally set if provided, the position in bytes from which to read the file, by default None
+    __lines_read : int, optional
+        Internal cumulative physical line count, including ignored lines.
     **parser_kwargs
-        arguments to pass to a custom tail parsing function if provided
+        arguments to pass to a custom tail parsing function if provided.
+        Custom parsers receive ``__line_numbers``, the 1-based physical line
+        positions of the lines in ``file_content``. Built-in parsers return
+        aligned positions in metadata so each callback receives ``line_number``.
 
     Returns
     -------
@@ -558,14 +536,22 @@ def record_log(
         * metadata outlining properties such as modified time etc.
         * actual recorded data from the file.
     """
+    if __read_bytes is None:
+        __lines_read = 0
     __read_bytes, _lines = tail_file_n_bytes(input_file, __read_bytes)
+    _numbered_lines = list(enumerate(_lines, start=__lines_read + 1))
+    __lines_read += len(_lines)
+    # Always preserve read state, including blocks containing only skipped lines.
+    _read_state = {"__read_bytes": __read_bytes, "__lines_read": __lines_read}
+    parser_kwargs.pop("__line_numbers", None)
+    parser_kwargs.pop("line_number", None)
 
     # Check if there are patterns defined for lines that should be ignored
     # if this is the case loop through all patterns for each line,
     # the patterns can either be string literals or regex compiled patterns
     if ignore_lines:
-        _passing_lines: list[str] = []
-        for line in _lines:
+        _passing_lines: list[tuple[int, str]] = []
+        for line_number, line in _numbered_lines:
             if any(
                 [
                     any(
@@ -578,8 +564,11 @@ def record_log(
                 ]
             ):
                 continue
-            _passing_lines.append(line)
-        _lines = _passing_lines
+            _passing_lines.append((line_number, line))
+        _numbered_lines = _passing_lines
+
+    _line_numbers = [number for number, _ in _numbered_lines]
+    _lines = [line for _, line in _numbered_lines]
 
     if parser_func:
         # In general parser functions are assumed to parse blocks of information
@@ -589,10 +578,12 @@ def record_log(
             file_content="".join(_lines),  # type: ignore[call-arg]
             __input_file=input_file,
             __read_bytes=__read_bytes,
+            __line_numbers=_line_numbers,
             convert=convert,
             **parser_kwargs,
         )
-        return _parsed_content
+        _custom_metadata, _custom_data = _parsed_content
+        return _custom_metadata | _read_state, _custom_data
 
     _data: list[dict[str, typing.Any]] = []
     _metadata: dict[str, typing.Any] = {}
@@ -609,7 +600,7 @@ def record_log(
         _data.append(_processed)  # type: ignore
         _metadata = _metadata or _metadata_line
 
-    return _metadata, _data
+    return _metadata | _read_state | {"__line_numbers": _line_numbers}, _data
 
 
 # Built in parsers do not need to be validated by the File Monitor
diff --git a/multiparser/thread.py b/multiparser/thread.py
index 0d9dfd3..6980c25 100644
--- a/multiparser/thread.py
+++ b/multiparser/thread.py
@@ -117,8 +117,19 @@ def _prepare_parsed_data(
             return
         if isinstance(parsed_data[1][0], dict):
             _metadata, _data = parsed_data
-            for d in _data:
-                yield (_metadata, d)
+            _line_numbers = _metadata.get("__line_numbers")
+            if _line_numbers is not None and len(_line_numbers) != len(_data):
+                raise RuntimeError("Line numbers must align with parsed rows")
+            _shared_metadata = {
+                key: value
+                for key, value in _metadata.items()
+                if key != "__line_numbers"
+            }
+            for index, d in enumerate(_data):
+                _row_metadata = _shared_metadata.copy()
+                if _line_numbers is not None:
+                    _row_metadata["line_number"] = _line_numbers[index]
+                yield (_row_metadata, d)
             return
     elif (
         isinstance(parsed_data, tuple)
@@ -192,7 +203,13 @@ def _reparse_action(
     if not _parsed:
         return cached_metadata
 
-    _cached_metadata = cached_metadata
+    # Read state must advance even when a block produces no data rows.
+    _cached_metadata = {
+        key: value
+        for key, value in cached_metadata.items()
+        if key not in ("__line_numbers", "line_number")
+    } | _parsed[0]
+    _parsed = (_cached_metadata, _parsed[1])
 
     # If the parser method records a list of dictionaries as data
     # we need to ensure these are handled in the same way as for parsers
diff --git a/multiparser/typing.py b/multiparser/typing.py
index 2be8c17..21d9e1c 100644
--- a/multiparser/typing.py
+++ b/multiparser/typing.py
@@ -28,7 +28,7 @@
 TrackedValues = list[re.Pattern[str] | str] | re.Pattern[str] | str
 
 TimeStampedData = tuple[
-    dict[str, str | int | list[str]],
+    dict[str, str | int | list[str] | list[int]],
     dict[str, typing.Any] | list[dict[str, typing.Any]],
 ]
 
diff --git a/tests/test_data_parsing.py b/tests/test_data_parsing.py
index ec50ab4..5363486 100644
--- a/tests/test_data_parsing.py
+++ b/tests/test_data_parsing.py
@@ -3,7 +3,6 @@
 import string
 import tempfile
 import time
-import typing
 import os.path
 import importlib.util
 
@@ -25,6 +24,8 @@
     _extract_label_value_pair,
 )
 
+from multiparser.thread import _prepare_parsed_data, _reparse_action
+
 DATA_LIBRARY: str = os.path.join(os.path.dirname(__file__), "data")
 
 
@@ -264,3 +265,111 @@ def test_label_value_extraction(regex_result: tuple, label: str | None) -> None:
             tracked_val=re.compile("undefined"),
             type_descriptor="NoType",
         )
+
+
+@pytest.mark.parsing
+@pytest.mark.parametrize("parser", [log_record_csv, record_with_delimiter, None])
+def test_callback_line_numbers_across_batches(tmp_path, parser):
+    path = tmp_path / "data.csv"
+    path.write_text("")
+    captured = []
+    metadata = {}
+    kwargs = {"ignore_lines": [re.compile(r"^#")]}
+    if parser is record_with_delimiter:
+        kwargs["delimiter"] = ","
+
+    def parse():
+        nonlocal metadata
+        metadata = _reparse_action(
+            file_name=str(path),
+            file_type=None,
+            cached_metadata=metadata,
+            modified_time="",
+            tracked_vals=[("value", re.compile(r"^(\d+)$"))],
+            parsing_callback=mp_parse.record_log,
+            parser_kwargs=kwargs,
+            cstm_parser=parser,
+            lock=None,
+            monitor_callback=lambda data, meta: captured.append((data, meta)),
+            flatten_data=False,
+        )
+
+    parse()  # An initially empty file has no header yet.
+    assert captured == []
+    assert metadata["__lines_read"] == 0
+
+    path.write_text("\n# ignored before header\n")
+    parse()
+    assert captured == []
+    assert metadata["__lines_read"] == 2
+
+    with path.open("a") as stream:
+        stream.write("value\n\n10\n# ignored\n20\n\n")
+    parse()
+    assert [meta["line_number"] for _, meta in captured] == [5, 7]
+    assert [data for data, _ in captured] == [{"value": 10}, {"value": 20}]
+    assert metadata["__lines_read"] == 8
+    assert "__read_bytes" in metadata
+    assert all("__line_numbers" not in meta for _, meta in captured)
+
+    # No callbacks, but both ignored and blank lines must advance the cursor.
+    with path.open("a") as stream:
+        stream.write("# ignored again\n\n")
+    parse()
+    assert len(captured) == 2
+    assert metadata["__lines_read"] == 10
+
+    with path.open("a") as stream:
+        stream.write("30\n\n40\n")
+    parse()
+    assert [meta["line_number"] for _, meta in captured] == [5, 7, 11, 13]
+    assert captured[-1][0] == {"value": 40}
+    assert metadata["__lines_read"] == 13
+    # Each callback gets independent metadata, not an overwritten shared dict.
+    assert captured[0][1]["line_number"] == 5
+
+    parse()  # No new content must not repeat any rows or lose state.
+    assert len(captured) == 4
+    assert metadata["__lines_read"] == 13
+
+
+@pytest.mark.parsing
+@pytest.mark.parametrize("headers", [None, ["value"]])
+def test_direct_csv_parser_line_numbers(tmp_path, headers):
+    path = tmp_path / "data.csv"
+    path.write_text("value\n\n1\n2\n")
+    parsed = log_record_csv(
+        file_content=path.read_text(),
+        headers=headers,
+        __input_file=str(path),
+        __read_bytes=path.stat().st_size,
+    )
+    rows = [
+        (meta["line_number"], data)
+        for meta, data in _prepare_parsed_data(parsed)
+        if data
+    ]
+    assert rows == [(3, {"value": 1}), (4, {"value": 2})]
+
+
+@pytest.mark.parsing
+def test_legacy_custom_parser_does_not_get_inferred_row_positions(tmp_path):
+    path = tmp_path / "data.log"
+    path.write_text("first\nsecond\n")
+
+    def parser(file_content, **kwargs):
+        assert kwargs["__line_numbers"] == [1, 2]
+        return {"custom": "metadata"}, [{"block": file_content}]
+
+    metadata, data = mp_parse.record_log(str(path), parser_func=parser)
+    assert metadata["__lines_read"] == 2
+    assert metadata["__read_bytes"] == path.stat().st_size
+    assert "line_number" not in list(_prepare_parsed_data((metadata, data)))[0][0]
+
+
+@pytest.mark.parsing
+@pytest.mark.parametrize("data", [{"x": 1}, [{"x": 1}, {"x": 2}], []])
+def test_legacy_parser_return_formats(data):
+    rows = list(_prepare_parsed_data(({"custom": "metadata"}, data)))
+    expected = data if isinstance(data, list) else [data]
+    assert rows == [({"custom": "metadata"}, row) for row in expected]

From 5056d09e8f3f90f854a226d51e92c909d9375568 Mon Sep 17 00:00:00 2001
From: Matt <matthew.field@ukaea.uk>
Date: Fri, 30 Oct 2026 14:38:59 +0000
Subject: [PATCH 2/8] Change to not include skipped lines

---
 docs/custom_parsers.md      | 23 ++++++--------
 docs/tailing.md             | 15 ++++-----
 multiparser/parsing/tail.py | 62 +++++++++++++++++++++++--------------
 tests/test_data_parsing.py  | 55 ++++++++++++++++++++++----------
 4 files changed, 94 insertions(+), 61 deletions(-)

diff --git a/docs/custom_parsers.md b/docs/custom_parsers.md
index aa6bede..651337e 100644
--- a/docs/custom_parsers.md
+++ b/docs/custom_parsers.md
@@ -78,19 +78,16 @@ with multiparser.FileMonitor(timeout=10) as monitor:
 In the case where you would like your parser function to accept additional keyword arguments you can add these
 to the definition and pass them to tracking using the `parser_kwargs` argument.
 
-### Line positions
-
-Tail parsers receive `__line_numbers` through their keyword arguments: a list of
-1-based physical file line numbers corresponding to the lines in `file_content`.
-Lines excluded by `skip_lines_w_pattern` have already been removed, but their
-positions still count, so the list may contain gaps.
-
-If a custom parser returns a list of row dictionaries, it can include
-`"__line_numbers": [...]` in its returned metadata to associate a position with
-each output row. The list must align with the **returned rows**, not necessarily
-all input lines. Each callback then receives its row's `metadata["line_number"]`.
-Existing custom parsers can ignore this feature and retain their current return
-format. Internal read-state metadata is maintained by Multiparser.
+### Parsed row numbers
+
+Each tail callback receives `metadata["line_number"]`: the 1-based number of the
+parsed data row, continuing across appended batches for that file. A non-empty
+returned dictionary counts as one row; in a returned list, each non-empty
+dictionary counts as one row. Empty dictionaries and empty lists do not count.
+This is independent of the number of input lines consumed, so a block parser
+returning one dictionary produces one numbered row even if the block spans
+several file lines. Multiparser maintains numbering and read state automatically;
+custom parsers do not need to change their return format.
 
 ## Class methods as parsers
 
diff --git a/docs/tailing.md b/docs/tailing.md
index a7499d8..5b847eb 100644
--- a/docs/tailing.md
+++ b/docs/tailing.md
@@ -77,10 +77,11 @@ Default: `None`
 
 Optional callback exclusively for this file set, this overwrites the global callback (if specified) within the `FileMonitor` definition (see [here](./file_monitor.md#per-thread-callback)).
 
-The built-in tail parsers (regular-expression logs, CSV, and other delimited files)
-include `metadata["line_number"]` in each callback. This is the **1-based physical
-file line**, counting headers, empty lines, and lines excluded by
-`skip_lines_w_pattern`. Numbering continues across appended batches and is
+Tail parsers include `metadata["line_number"]` in each callback. This is the
+**1-based parsed data row number**, not the physical file line number. The first
+non-empty data dictionary returned by the parser is row 1, the next is row 2,
+and so on. Headers, blank lines, ignored lines, and anything else that produces
+no data do not count. Numbering continues across appended batches and is
 independent for each file. As with tailing generally, this assumes append-only
 files with complete, newline-terminated records; CSV records spanning multiple
 physical lines are not supported.
@@ -100,9 +101,9 @@ monitor.tail(
 )
 ```
 
-Custom block parsers are not assigned line numbers automatically, since an output
-row may represent several input lines. See [custom parsers](./custom_parsers.md#log-parsers)
-for explicit row-position support.
+Custom parsers use the same numbering: a returned dictionary counts as one row,
+and each non-empty dictionary in a returned list counts as one row, regardless
+of how many input lines were used to produce it.
 
 
 ### `parser_func`
diff --git a/multiparser/parsing/tail.py b/multiparser/parsing/tail.py
index e6fe168..55ac56f 100644
--- a/multiparser/parsing/tail.py
+++ b/multiparser/parsing/tail.py
@@ -228,6 +228,24 @@ def _record_any_delimited(
     return {}, _out_filtered
 
 
+def _number_parsed_rows(parsed: TimeStampedData, rows_read: int = 0) -> TimeStampedData:
+    """Number non-empty parsed rows, independently of input file lines."""
+    metadata, data = parsed
+    metadata = metadata.copy()
+    if isinstance(data, list):
+        numbers = []
+        for row in data:
+            if row:
+                rows_read += 1
+            numbers.append(rows_read)
+        metadata["__line_numbers"] = numbers
+    elif data:
+        rows_read += 1
+        metadata["line_number"] = rows_read
+    metadata["__lines_read"] = rows_read
+    return metadata, data
+
+
 def record_with_delimiter(
     file_content: str,
     delimiter: str,
@@ -260,10 +278,10 @@ def record_with_delimiter(
     # The delimiter parser assumes each line is a new data entry so
     # revert back to list of lines here
     _file_lines = file_content.split("\n")
-    _line_numbers = kwargs.pop("__line_numbers", list(range(1, len(_file_lines) + 1)))
+    _rows_read = kwargs.pop("__lines_read", 0)
 
     _parsed_data: tuple[dict[str, typing.Any], list[dict[str, typing.Any]]] = (
-        {"__line_numbers": []},
+        {},
         [],
     )
 
@@ -271,12 +289,11 @@ def record_with_delimiter(
         _parsed_data[0]["headers"] = headers
 
     if not any(_file_lines):
-        return {}, []
+        return _number_parsed_rows(_parsed_data, _rows_read)
 
-    for line_number, file_line in zip(_line_numbers, _file_lines):
+    for file_line in _file_lines:
         if not file_line:
             continue
-        _parsed_data[0]["__line_numbers"].append(line_number)
         _parsed_line: TimeStampedData = _record_any_delimited(  # type: ignore[call-arg]
             file_content=file_line,
             delimiter=delimiter,
@@ -306,7 +323,7 @@ def record_with_delimiter(
     if not _parsed_data[0].get("headers"):
         raise AssertionError("Failed to retrieve file header during initial read")
 
-    return _parsed_data
+    return _number_parsed_rows(_parsed_data, _rows_read)
 
 
 def record_csv(
@@ -523,12 +540,11 @@ def record_log(
     __read_bytes : int | None, optional
         internally set if provided, the position in bytes from which to read the file, by default None
     __lines_read : int, optional
-        Internal cumulative physical line count, including ignored lines.
+        Internal cumulative count of non-empty parsed data rows.
     **parser_kwargs
         arguments to pass to a custom tail parsing function if provided.
-        Custom parsers receive ``__line_numbers``, the 1-based physical line
-        positions of the lines in ``file_content``. Built-in parsers return
-        aligned positions in metadata so each callback receives ``line_number``.
+        Each non-empty returned data row receives a consecutive 1-based
+        ``line_number`` in callback metadata, continuing across read batches.
 
     Returns
     -------
@@ -539,10 +555,6 @@ def record_log(
     if __read_bytes is None:
         __lines_read = 0
     __read_bytes, _lines = tail_file_n_bytes(input_file, __read_bytes)
-    _numbered_lines = list(enumerate(_lines, start=__lines_read + 1))
-    __lines_read += len(_lines)
-    # Always preserve read state, including blocks containing only skipped lines.
-    _read_state = {"__read_bytes": __read_bytes, "__lines_read": __lines_read}
     parser_kwargs.pop("__line_numbers", None)
     parser_kwargs.pop("line_number", None)
 
@@ -550,8 +562,8 @@ def record_log(
     # if this is the case loop through all patterns for each line,
     # the patterns can either be string literals or regex compiled patterns
     if ignore_lines:
-        _passing_lines: list[tuple[int, str]] = []
-        for line_number, line in _numbered_lines:
+        _passing_lines: list[str] = []
+        for line in _lines:
             if any(
                 [
                     any(
@@ -564,11 +576,8 @@ def record_log(
                 ]
             ):
                 continue
-            _passing_lines.append((line_number, line))
-        _numbered_lines = _passing_lines
-
-    _line_numbers = [number for number, _ in _numbered_lines]
-    _lines = [line for _, line in _numbered_lines]
+            _passing_lines.append(line)
+        _lines = _passing_lines
 
     if parser_func:
         # In general parser functions are assumed to parse blocks of information
@@ -578,12 +587,15 @@ def record_log(
             file_content="".join(_lines),  # type: ignore[call-arg]
             __input_file=input_file,
             __read_bytes=__read_bytes,
-            __line_numbers=_line_numbers,
+            __lines_read=__lines_read,
             convert=convert,
             **parser_kwargs,
         )
         _custom_metadata, _custom_data = _parsed_content
-        return _custom_metadata | _read_state, _custom_data
+        return _number_parsed_rows(
+            (_custom_metadata | {"__read_bytes": __read_bytes}, _custom_data),
+            __lines_read,
+        )
 
     _data: list[dict[str, typing.Any]] = []
     _metadata: dict[str, typing.Any] = {}
@@ -600,7 +612,9 @@ def record_log(
         _data.append(_processed)  # type: ignore
         _metadata = _metadata or _metadata_line
 
-    return _metadata | _read_state | {"__line_numbers": _line_numbers}, _data
+    return _number_parsed_rows(
+        (_metadata | {"__read_bytes": __read_bytes}, _data), __lines_read
+    )
 
 
 # Built in parsers do not need to be validated by the File Monitor
diff --git a/tests/test_data_parsing.py b/tests/test_data_parsing.py
index 5363486..02f7cd5 100644
--- a/tests/test_data_parsing.py
+++ b/tests/test_data_parsing.py
@@ -301,43 +301,52 @@ def parse():
     path.write_text("\n# ignored before header\n")
     parse()
     assert captured == []
-    assert metadata["__lines_read"] == 2
+    assert metadata["__lines_read"] == 0
 
     with path.open("a") as stream:
         stream.write("value\n\n10\n# ignored\n20\n\n")
     parse()
-    assert [meta["line_number"] for _, meta in captured] == [5, 7]
+    assert [meta["line_number"] for _, meta in captured] == [1, 2]
     assert [data for data, _ in captured] == [{"value": 10}, {"value": 20}]
-    assert metadata["__lines_read"] == 8
+    assert metadata["__lines_read"] == 2
     assert "__read_bytes" in metadata
     assert all("__line_numbers" not in meta for _, meta in captured)
 
-    # No callbacks, but both ignored and blank lines must advance the cursor.
+    # An ignored-only batch advances the byte cursor, not the line count.
+    previous_bytes = metadata["__read_bytes"]
+    with path.open("a") as stream:
+        stream.write("# entirely ignored batch\n")
+    parse()
+    assert len(captured) == 2
+    assert metadata["__lines_read"] == 2
+    assert metadata["__read_bytes"] > previous_bytes
+
+    # Ignored lines advance only the byte cursor; blank lines do not count either.
     with path.open("a") as stream:
         stream.write("# ignored again\n\n")
     parse()
     assert len(captured) == 2
-    assert metadata["__lines_read"] == 10
+    assert metadata["__lines_read"] == 2
 
     with path.open("a") as stream:
         stream.write("30\n\n40\n")
     parse()
-    assert [meta["line_number"] for _, meta in captured] == [5, 7, 11, 13]
+    assert [meta["line_number"] for _, meta in captured] == [1, 2, 3, 4]
     assert captured[-1][0] == {"value": 40}
-    assert metadata["__lines_read"] == 13
+    assert metadata["__lines_read"] == 4
     # Each callback gets independent metadata, not an overwritten shared dict.
-    assert captured[0][1]["line_number"] == 5
+    assert captured[0][1]["line_number"] == 1
 
     parse()  # No new content must not repeat any rows or lose state.
     assert len(captured) == 4
-    assert metadata["__lines_read"] == 13
+    assert metadata["__lines_read"] == 4
 
 
 @pytest.mark.parsing
-@pytest.mark.parametrize("headers", [None, ["value"]])
+@pytest.mark.parametrize("headers", [None, ["x", "y"]])
 def test_direct_csv_parser_line_numbers(tmp_path, headers):
     path = tmp_path / "data.csv"
-    path.write_text("value\n\n1\n2\n")
+    path.write_text("x,y\n\n1,2\n3,4\n")
     parsed = log_record_csv(
         file_content=path.read_text(),
         headers=headers,
@@ -349,22 +358,34 @@ def test_direct_csv_parser_line_numbers(tmp_path, headers):
         for meta, data in _prepare_parsed_data(parsed)
         if data
     ]
-    assert rows == [(3, {"value": 1}), (4, {"value": 2})]
+    assert rows == [(1, {"x": 1, "y": 2}), (2, {"x": 3, "y": 4})]
 
 
 @pytest.mark.parsing
-def test_legacy_custom_parser_does_not_get_inferred_row_positions(tmp_path):
+@pytest.mark.parametrize("single_row", [False, True])
+def test_custom_parser_numbers_returned_rows(tmp_path, single_row):
     path = tmp_path / "data.log"
     path.write_text("first\nsecond\n")
 
     def parser(file_content, **kwargs):
-        assert kwargs["__line_numbers"] == [1, 2]
-        return {"custom": "metadata"}, [{"block": file_content}]
+        row = {"block": file_content}
+        return {"custom": "metadata"}, row if single_row else [{}, row, {}]
 
     metadata, data = mp_parse.record_log(str(path), parser_func=parser)
-    assert metadata["__lines_read"] == 2
+    assert metadata["__lines_read"] == 1
     assert metadata["__read_bytes"] == path.stat().st_size
-    assert "line_number" not in list(_prepare_parsed_data((metadata, data)))[0][0]
+    rows = [(meta, row) for meta, row in _prepare_parsed_data((metadata, data)) if row]
+    assert len(rows) == 1
+    assert rows[0][0]["line_number"] == 1
+    assert rows[0][0]["custom"] == "metadata"
+
+    with path.open("a") as stream:
+        stream.write("third\nfourth\n")
+    metadata, data = mp_parse.record_log(str(path), parser_func=parser, **metadata)
+    rows = [(meta, row) for meta, row in _prepare_parsed_data((metadata, data)) if row]
+    assert len(rows) == 1
+    assert rows[0][0]["line_number"] == 2
+    assert metadata["__lines_read"] == 2
 
 
 @pytest.mark.parsing

From 1730daeb7c0722e15f894d3ea2236ff63e743fe1 Mon Sep 17 00:00:00 2001
From: Matt <matthew.field@ukaea.uk>
Date: Fri, 30 Oct 2026 15:43:56 +0000
Subject: [PATCH 3/8] Refine docs

---
 docs/custom_parsers.md | 11 -----------
 docs/tailing.md        | 39 ++++++++++++++++++---------------------
 2 files changed, 18 insertions(+), 32 deletions(-)

diff --git a/docs/custom_parsers.md b/docs/custom_parsers.md
index 651337e..0a0073d 100644
--- a/docs/custom_parsers.md
+++ b/docs/custom_parsers.md
@@ -78,17 +78,6 @@ with multiparser.FileMonitor(timeout=10) as monitor:
 In the case where you would like your parser function to accept additional keyword arguments you can add these
 to the definition and pass them to tracking using the `parser_kwargs` argument.
 
-### Parsed row numbers
-
-Each tail callback receives `metadata["line_number"]`: the 1-based number of the
-parsed data row, continuing across appended batches for that file. A non-empty
-returned dictionary counts as one row; in a returned list, each non-empty
-dictionary counts as one row. Empty dictionaries and empty lists do not count.
-This is independent of the number of input lines consumed, so a block parser
-returning one dictionary produces one numbered row even if the block spans
-several file lines. Multiparser maintains numbering and read state automatically;
-custom parsers do not need to change their return format.
-
 ## Class methods as parsers
 
 When custom parsers are decorated any positional and keyword arguments are propagated through, i.e.:
diff --git a/docs/tailing.md b/docs/tailing.md
index 5b847eb..afa5996 100644
--- a/docs/tailing.md
+++ b/docs/tailing.md
@@ -77,12 +77,28 @@ Default: `None`
 
 Optional callback exclusively for this file set, this overwrites the global callback (if specified) within the `FileMonitor` definition (see [here](./file_monitor.md#per-thread-callback)).
 
+
+### `parser_func`
+`#!python Callable[[str, *_], tuple[dict[str, Any], dict[str, Any]]] | None`
+
+Default: `None`
+
+Specific a custom parser for this file set, this must be a log parsing function wrapped in the `multiparser.parsing.tail.log_parser` decorator, for more information see [here](./custom_parsers.md#log-parsers).
+
+### `parser_kwargs`
+`#!python dict[str, Any] | None`
+
+Default: `None`
+
+Keyword arguments to pass to the function defined above, useful if you want to specify near-identical parsing of multiple file sets with slight customisation.
+
+## Metadata
+
 Tail parsers include `metadata["line_number"]` in each callback. This is the
 **1-based parsed data row number**, not the physical file line number. The first
 non-empty data dictionary returned by the parser is row 1, the next is row 2,
 and so on. Headers, blank lines, ignored lines, and anything else that produces
-no data do not count. Numbering continues across appended batches and is
-independent for each file. As with tailing generally, this assumes append-only
+no data do not count. As with tailing generally, this assumes append-only
 files with complete, newline-terminated records; CSV records spanning multiple
 physical lines are not supported.
 
@@ -101,25 +117,6 @@ monitor.tail(
 )
 ```
 
-Custom parsers use the same numbering: a returned dictionary counts as one row,
-and each non-empty dictionary in a returned list counts as one row, regardless
-of how many input lines were used to produce it.
-
-
-### `parser_func`
-`#!python Callable[[str, *_], tuple[dict[str, Any], dict[str, Any]]] | None`
-
-Default: `None`
-
-Specific a custom parser for this file set, this must be a log parsing function wrapped in the `multiparser.parsing.tail.log_parser` decorator, for more information see [here](./custom_parsers.md#log-parsers).
-
-### `parser_kwargs`
-`#!python dict[str, Any] | None`
-
-Default: `None`
-
-Keyword arguments to pass to the function defined above, useful if you want to specify near-identical parsing of multiple file sets with slight customisation.
-
 ## Returns
 
 `#!python None`

From 62eafc57a205fd1647cc6c3096df7bf593636f93 Mon Sep 17 00:00:00 2001
From: Matt <matthew.field@ukaea.uk>
Date: Fri, 30 Oct 2026 15:52:25 +0000
Subject: [PATCH 4/8] Use an int instead of list of ints

---
 multiparser/parsing/tail.py |  8 +-------
 multiparser/thread.py       | 25 ++++++++++---------------
 multiparser/typing.py       |  2 +-
 tests/test_data_parsing.py  |  4 +++-
 4 files changed, 15 insertions(+), 24 deletions(-)

diff --git a/multiparser/parsing/tail.py b/multiparser/parsing/tail.py
index 55ac56f..e75e39e 100644
--- a/multiparser/parsing/tail.py
+++ b/multiparser/parsing/tail.py
@@ -233,12 +233,7 @@ def _number_parsed_rows(parsed: TimeStampedData, rows_read: int = 0) -> TimeStam
     metadata, data = parsed
     metadata = metadata.copy()
     if isinstance(data, list):
-        numbers = []
-        for row in data:
-            if row:
-                rows_read += 1
-            numbers.append(rows_read)
-        metadata["__line_numbers"] = numbers
+        rows_read += sum(bool(row) for row in data)
     elif data:
         rows_read += 1
         metadata["line_number"] = rows_read
@@ -555,7 +550,6 @@ def record_log(
     if __read_bytes is None:
         __lines_read = 0
     __read_bytes, _lines = tail_file_n_bytes(input_file, __read_bytes)
-    parser_kwargs.pop("__line_numbers", None)
     parser_kwargs.pop("line_number", None)
 
     # Check if there are patterns defined for lines that should be ignored
diff --git a/multiparser/thread.py b/multiparser/thread.py
index 6980c25..d651b50 100644
--- a/multiparser/thread.py
+++ b/multiparser/thread.py
@@ -117,18 +117,15 @@ def _prepare_parsed_data(
             return
         if isinstance(parsed_data[1][0], dict):
             _metadata, _data = parsed_data
-            _line_numbers = _metadata.get("__line_numbers")
-            if _line_numbers is not None and len(_line_numbers) != len(_data):
-                raise RuntimeError("Line numbers must align with parsed rows")
-            _shared_metadata = {
-                key: value
-                for key, value in _metadata.items()
-                if key != "__line_numbers"
-            }
-            for index, d in enumerate(_data):
-                _row_metadata = _shared_metadata.copy()
-                if _line_numbers is not None:
-                    _row_metadata["line_number"] = _line_numbers[index]
+            _row_number = _metadata.get("__lines_read")
+            if _row_number is not None:
+                # The cached total includes this batch. Start at the previous total.
+                _row_number -= sum(bool(row) for row in _data)
+            for d in _data:
+                _row_metadata = _metadata.copy()
+                if _row_number is not None and d:
+                    _row_number += 1
+                    _row_metadata["line_number"] = _row_number
                 yield (_row_metadata, d)
             return
     elif (
@@ -205,9 +202,7 @@ def _reparse_action(
 
     # Read state must advance even when a block produces no data rows.
     _cached_metadata = {
-        key: value
-        for key, value in cached_metadata.items()
-        if key not in ("__line_numbers", "line_number")
+        key: value for key, value in cached_metadata.items() if key != "line_number"
     } | _parsed[0]
     _parsed = (_cached_metadata, _parsed[1])
 
diff --git a/multiparser/typing.py b/multiparser/typing.py
index 21d9e1c..2be8c17 100644
--- a/multiparser/typing.py
+++ b/multiparser/typing.py
@@ -28,7 +28,7 @@
 TrackedValues = list[re.Pattern[str] | str] | re.Pattern[str] | str
 
 TimeStampedData = tuple[
-    dict[str, str | int | list[str] | list[int]],
+    dict[str, str | int | list[str]],
     dict[str, typing.Any] | list[dict[str, typing.Any]],
 ]
 
diff --git a/tests/test_data_parsing.py b/tests/test_data_parsing.py
index 02f7cd5..f5f9f42 100644
--- a/tests/test_data_parsing.py
+++ b/tests/test_data_parsing.py
@@ -310,7 +310,7 @@ def parse():
     assert [data for data, _ in captured] == [{"value": 10}, {"value": 20}]
     assert metadata["__lines_read"] == 2
     assert "__read_bytes" in metadata
-    assert all("__line_numbers" not in meta for _, meta in captured)
+    assert isinstance(metadata["__lines_read"], int)
 
     # An ignored-only batch advances the byte cursor, not the line count.
     previous_bytes = metadata["__read_bytes"]
@@ -353,6 +353,8 @@ def test_direct_csv_parser_line_numbers(tmp_path, headers):
         __input_file=str(path),
         __read_bytes=path.stat().st_size,
     )
+    assert isinstance(parsed[0]["__lines_read"], int)
+    assert "__line_numbers" not in parsed[0]
     rows = [
         (meta["line_number"], data)
         for meta, data in _prepare_parsed_data(parsed)

From 0911ad426f2b9deedbec1831a251488eac381208 Mon Sep 17 00:00:00 2001
From: Matt <matthew.field@ukaea.uk>
Date: Fri, 30 Oct 2026 15:59:42 +0000
Subject: [PATCH 5/8] Rename

---
 docs/tailing.md             |  4 ++--
 multiparser/parsing/tail.py | 22 ++++++++++----------
 multiparser/thread.py       |  6 +++---
 tests/test_data_parsing.py  | 40 ++++++++++++++++++-------------------
 4 files changed, 36 insertions(+), 36 deletions(-)

diff --git a/docs/tailing.md b/docs/tailing.md
index afa5996..2bcb636 100644
--- a/docs/tailing.md
+++ b/docs/tailing.md
@@ -94,7 +94,7 @@ Keyword arguments to pass to the function defined above, useful if you want to s
 
 ## Metadata
 
-Tail parsers include `metadata["line_number"]` in each callback. This is the
+Tail parsers include `metadata["row_number"]` in each callback. This is the
 **1-based parsed data row number**, not the physical file line number. The first
 non-empty data dictionary returned by the parser is row 1, the next is row 2,
 and so on. Headers, blank lines, ignored lines, and anything else that produces
@@ -107,7 +107,7 @@ from multiparser.parsing.tail import record_csv
 
 
 def on_row(data, metadata):
-    print(metadata["file_name"], metadata["line_number"], data)
+    print(metadata["file_name"], metadata["row_number"], data)
 
 
 monitor.tail(
diff --git a/multiparser/parsing/tail.py b/multiparser/parsing/tail.py
index e75e39e..ec4d6bf 100644
--- a/multiparser/parsing/tail.py
+++ b/multiparser/parsing/tail.py
@@ -236,8 +236,8 @@ def _number_parsed_rows(parsed: TimeStampedData, rows_read: int = 0) -> TimeStam
         rows_read += sum(bool(row) for row in data)
     elif data:
         rows_read += 1
-        metadata["line_number"] = rows_read
-    metadata["__lines_read"] = rows_read
+        metadata["row_number"] = rows_read
+    metadata["__rows_read"] = rows_read
     return metadata, data
 
 
@@ -273,7 +273,7 @@ def record_with_delimiter(
     # The delimiter parser assumes each line is a new data entry so
     # revert back to list of lines here
     _file_lines = file_content.split("\n")
-    _rows_read = kwargs.pop("__lines_read", 0)
+    _rows_read = kwargs.pop("__rows_read", 0)
 
     _parsed_data: tuple[dict[str, typing.Any], list[dict[str, typing.Any]]] = (
         {},
@@ -510,7 +510,7 @@ def record_log(
     ignore_lines: list[re.Pattern[str]] | None = None,
     parser_func: ParserFunction | None = None,
     __read_bytes: int | None = None,
-    __lines_read: int = 0,
+    __rows_read: int = 0,
     **parser_kwargs,
 ) -> TimeStampedData:
     """Record lines within a log type file.
@@ -534,12 +534,12 @@ def record_log(
         specify an alternative tail parsing function
     __read_bytes : int | None, optional
         internally set if provided, the position in bytes from which to read the file, by default None
-    __lines_read : int, optional
+    __rows_read : int, optional
         Internal cumulative count of non-empty parsed data rows.
     **parser_kwargs
         arguments to pass to a custom tail parsing function if provided.
         Each non-empty returned data row receives a consecutive 1-based
-        ``line_number`` in callback metadata, continuing across read batches.
+        ``row_number`` in callback metadata, continuing across read batches.
 
     Returns
     -------
@@ -548,9 +548,9 @@ def record_log(
         * actual recorded data from the file.
     """
     if __read_bytes is None:
-        __lines_read = 0
+        __rows_read = 0
     __read_bytes, _lines = tail_file_n_bytes(input_file, __read_bytes)
-    parser_kwargs.pop("line_number", None)
+    parser_kwargs.pop("row_number", None)
 
     # Check if there are patterns defined for lines that should be ignored
     # if this is the case loop through all patterns for each line,
@@ -581,14 +581,14 @@ def record_log(
             file_content="".join(_lines),  # type: ignore[call-arg]
             __input_file=input_file,
             __read_bytes=__read_bytes,
-            __lines_read=__lines_read,
+            __rows_read=__rows_read,
             convert=convert,
             **parser_kwargs,
         )
         _custom_metadata, _custom_data = _parsed_content
         return _number_parsed_rows(
             (_custom_metadata | {"__read_bytes": __read_bytes}, _custom_data),
-            __lines_read,
+            __rows_read,
         )
 
     _data: list[dict[str, typing.Any]] = []
@@ -607,7 +607,7 @@ def record_log(
         _metadata = _metadata or _metadata_line
 
     return _number_parsed_rows(
-        (_metadata | {"__read_bytes": __read_bytes}, _data), __lines_read
+        (_metadata | {"__read_bytes": __read_bytes}, _data), __rows_read
     )
 
 
diff --git a/multiparser/thread.py b/multiparser/thread.py
index d651b50..723b74e 100644
--- a/multiparser/thread.py
+++ b/multiparser/thread.py
@@ -117,7 +117,7 @@ def _prepare_parsed_data(
             return
         if isinstance(parsed_data[1][0], dict):
             _metadata, _data = parsed_data
-            _row_number = _metadata.get("__lines_read")
+            _row_number = _metadata.get("__rows_read")
             if _row_number is not None:
                 # The cached total includes this batch. Start at the previous total.
                 _row_number -= sum(bool(row) for row in _data)
@@ -125,7 +125,7 @@ def _prepare_parsed_data(
                 _row_metadata = _metadata.copy()
                 if _row_number is not None and d:
                     _row_number += 1
-                    _row_metadata["line_number"] = _row_number
+                    _row_metadata["row_number"] = _row_number
                 yield (_row_metadata, d)
             return
     elif (
@@ -202,7 +202,7 @@ def _reparse_action(
 
     # Read state must advance even when a block produces no data rows.
     _cached_metadata = {
-        key: value for key, value in cached_metadata.items() if key != "line_number"
+        key: value for key, value in cached_metadata.items() if key != "row_number"
     } | _parsed[0]
     _parsed = (_cached_metadata, _parsed[1])
 
diff --git a/tests/test_data_parsing.py b/tests/test_data_parsing.py
index f5f9f42..844df96 100644
--- a/tests/test_data_parsing.py
+++ b/tests/test_data_parsing.py
@@ -269,7 +269,7 @@ def test_label_value_extraction(regex_result: tuple, label: str | None) -> None:
 
 @pytest.mark.parsing
 @pytest.mark.parametrize("parser", [log_record_csv, record_with_delimiter, None])
-def test_callback_line_numbers_across_batches(tmp_path, parser):
+def test_callback_row_numbers_across_batches(tmp_path, parser):
     path = tmp_path / "data.csv"
     path.write_text("")
     captured = []
@@ -296,21 +296,21 @@ def parse():
 
     parse()  # An initially empty file has no header yet.
     assert captured == []
-    assert metadata["__lines_read"] == 0
+    assert metadata["__rows_read"] == 0
 
     path.write_text("\n# ignored before header\n")
     parse()
     assert captured == []
-    assert metadata["__lines_read"] == 0
+    assert metadata["__rows_read"] == 0
 
     with path.open("a") as stream:
         stream.write("value\n\n10\n# ignored\n20\n\n")
     parse()
-    assert [meta["line_number"] for _, meta in captured] == [1, 2]
+    assert [meta["row_number"] for _, meta in captured] == [1, 2]
     assert [data for data, _ in captured] == [{"value": 10}, {"value": 20}]
-    assert metadata["__lines_read"] == 2
+    assert metadata["__rows_read"] == 2
     assert "__read_bytes" in metadata
-    assert isinstance(metadata["__lines_read"], int)
+    assert isinstance(metadata["__rows_read"], int)
 
     # An ignored-only batch advances the byte cursor, not the line count.
     previous_bytes = metadata["__read_bytes"]
@@ -318,7 +318,7 @@ def parse():
         stream.write("# entirely ignored batch\n")
     parse()
     assert len(captured) == 2
-    assert metadata["__lines_read"] == 2
+    assert metadata["__rows_read"] == 2
     assert metadata["__read_bytes"] > previous_bytes
 
     # Ignored lines advance only the byte cursor; blank lines do not count either.
@@ -326,25 +326,25 @@ def parse():
         stream.write("# ignored again\n\n")
     parse()
     assert len(captured) == 2
-    assert metadata["__lines_read"] == 2
+    assert metadata["__rows_read"] == 2
 
     with path.open("a") as stream:
         stream.write("30\n\n40\n")
     parse()
-    assert [meta["line_number"] for _, meta in captured] == [1, 2, 3, 4]
+    assert [meta["row_number"] for _, meta in captured] == [1, 2, 3, 4]
     assert captured[-1][0] == {"value": 40}
-    assert metadata["__lines_read"] == 4
+    assert metadata["__rows_read"] == 4
     # Each callback gets independent metadata, not an overwritten shared dict.
-    assert captured[0][1]["line_number"] == 1
+    assert captured[0][1]["row_number"] == 1
 
     parse()  # No new content must not repeat any rows or lose state.
     assert len(captured) == 4
-    assert metadata["__lines_read"] == 4
+    assert metadata["__rows_read"] == 4
 
 
 @pytest.mark.parsing
 @pytest.mark.parametrize("headers", [None, ["x", "y"]])
-def test_direct_csv_parser_line_numbers(tmp_path, headers):
+def test_direct_csv_parser_row_numbers(tmp_path, headers):
     path = tmp_path / "data.csv"
     path.write_text("x,y\n\n1,2\n3,4\n")
     parsed = log_record_csv(
@@ -353,10 +353,10 @@ def test_direct_csv_parser_line_numbers(tmp_path, headers):
         __input_file=str(path),
         __read_bytes=path.stat().st_size,
     )
-    assert isinstance(parsed[0]["__lines_read"], int)
-    assert "__line_numbers" not in parsed[0]
+    assert isinstance(parsed[0]["__rows_read"], int)
+    assert "__row_numbers" not in parsed[0]
     rows = [
-        (meta["line_number"], data)
+        (meta["row_number"], data)
         for meta, data in _prepare_parsed_data(parsed)
         if data
     ]
@@ -374,11 +374,11 @@ def parser(file_content, **kwargs):
         return {"custom": "metadata"}, row if single_row else [{}, row, {}]
 
     metadata, data = mp_parse.record_log(str(path), parser_func=parser)
-    assert metadata["__lines_read"] == 1
+    assert metadata["__rows_read"] == 1
     assert metadata["__read_bytes"] == path.stat().st_size
     rows = [(meta, row) for meta, row in _prepare_parsed_data((metadata, data)) if row]
     assert len(rows) == 1
-    assert rows[0][0]["line_number"] == 1
+    assert rows[0][0]["row_number"] == 1
     assert rows[0][0]["custom"] == "metadata"
 
     with path.open("a") as stream:
@@ -386,8 +386,8 @@ def parser(file_content, **kwargs):
     metadata, data = mp_parse.record_log(str(path), parser_func=parser, **metadata)
     rows = [(meta, row) for meta, row in _prepare_parsed_data((metadata, data)) if row]
     assert len(rows) == 1
-    assert rows[0][0]["line_number"] == 2
-    assert metadata["__lines_read"] == 2
+    assert rows[0][0]["row_number"] == 2
+    assert metadata["__rows_read"] == 2
 
 
 @pytest.mark.parsing

From 0a5d0c06c636b269a0eb05d7e52fe9b727ebaf96 Mon Sep 17 00:00:00 2001
From: Matt <matthew.field@ukaea.uk>
Date: Fri, 30 Oct 2026 16:01:20 +0000
Subject: [PATCH 6/8] Restore blank line filtering

---
 multiparser/parsing/tail.py | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/multiparser/parsing/tail.py b/multiparser/parsing/tail.py
index ec4d6bf..c160e61 100644
--- a/multiparser/parsing/tail.py
+++ b/multiparser/parsing/tail.py
@@ -272,7 +272,7 @@ def record_with_delimiter(
     """
     # The delimiter parser assumes each line is a new data entry so
     # revert back to list of lines here
-    _file_lines = file_content.split("\n")
+    _file_lines: list[str] = [i for i in file_content.split("\n") if i]
     _rows_read = kwargs.pop("__rows_read", 0)
 
     _parsed_data: tuple[dict[str, typing.Any], list[dict[str, typing.Any]]] = (
@@ -283,12 +283,10 @@ def record_with_delimiter(
     if headers:
         _parsed_data[0]["headers"] = headers
 
-    if not any(_file_lines):
+    if not _file_lines:
         return _number_parsed_rows(_parsed_data, _rows_read)
 
     for file_line in _file_lines:
-        if not file_line:
-            continue
         _parsed_line: TimeStampedData = _record_any_delimited(  # type: ignore[call-arg]
             file_content=file_line,
             delimiter=delimiter,

From 4557d4d0ab02e5af5f2d87a72d6bed4f992e8410 Mon Sep 17 00:00:00 2001
From: Matt <matthew.field@ukaea.uk>
Date: Fri, 30 Oct 2026 16:04:21 +0000
Subject: [PATCH 7/8] Docstring

---
 multiparser/parsing/tail.py | 17 ++++++++++++++++-
 1 file changed, 16 insertions(+), 1 deletion(-)

diff --git a/multiparser/parsing/tail.py b/multiparser/parsing/tail.py
index c160e61..24b3e6e 100644
--- a/multiparser/parsing/tail.py
+++ b/multiparser/parsing/tail.py
@@ -229,7 +229,22 @@ def _record_any_delimited(
 
 
 def _number_parsed_rows(parsed: TimeStampedData, rows_read: int = 0) -> TimeStampedData:
-    """Number non-empty parsed rows, independently of input file lines."""
+    """Number non-empty parsed rows, independently of input file lines.
+
+    Parameters
+    ----------
+    parsed: TimeStampedData
+        The data and metadata parsed from a file
+    rows_read: int, optional
+        The number of rows read up until this point, default is 0
+
+    Returns
+    -------
+    TimeStampedData
+        * metadata with row number information attached.
+        * actual recorded data from the file.
+
+    """
     metadata, data = parsed
     metadata = metadata.copy()
     if isinstance(data, list):

From 4dda8c6160583641eb2fab3f86c170459e6070ad Mon Sep 17 00:00:00 2001
From: Matt <matthew.field@ukaea.uk>
Date: Wed, 4 Nov 2026 13:23:09 +0000
Subject: [PATCH 8/8] Change to 0-based

---
 docs/tailing.md             |  4 ++--
 multiparser/parsing/tail.py |  4 ++--
 multiparser/thread.py       |  2 +-
 tests/test_data_parsing.py  | 14 +++++++-------
 4 files changed, 12 insertions(+), 12 deletions(-)

diff --git a/docs/tailing.md b/docs/tailing.md
index 2bcb636..f76cb0e 100644
--- a/docs/tailing.md
+++ b/docs/tailing.md
@@ -95,8 +95,8 @@ Keyword arguments to pass to the function defined above, useful if you want to s
 ## Metadata
 
 Tail parsers include `metadata["row_number"]` in each callback. This is the
-**1-based parsed data row number**, not the physical file line number. The first
-non-empty data dictionary returned by the parser is row 1, the next is row 2,
+**0-based parsed data row number**, not the physical file line number. The first
+non-empty data dictionary returned by the parser is row 0, the next is row 1,
 and so on. Headers, blank lines, ignored lines, and anything else that produces
 no data do not count. As with tailing generally, this assumes append-only
 files with complete, newline-terminated records; CSV records spanning multiple
diff --git a/multiparser/parsing/tail.py b/multiparser/parsing/tail.py
index 24b3e6e..e68e303 100644
--- a/multiparser/parsing/tail.py
+++ b/multiparser/parsing/tail.py
@@ -250,8 +250,8 @@ def _number_parsed_rows(parsed: TimeStampedData, rows_read: int = 0) -> TimeStam
     if isinstance(data, list):
         rows_read += sum(bool(row) for row in data)
     elif data:
-        rows_read += 1
         metadata["row_number"] = rows_read
+        rows_read += 1
     metadata["__rows_read"] = rows_read
     return metadata, data
 
@@ -551,7 +551,7 @@ def record_log(
         Internal cumulative count of non-empty parsed data rows.
     **parser_kwargs
         arguments to pass to a custom tail parsing function if provided.
-        Each non-empty returned data row receives a consecutive 1-based
+        Each non-empty returned data row receives a consecutive 0-based
         ``row_number`` in callback metadata, continuing across read batches.
 
     Returns
diff --git a/multiparser/thread.py b/multiparser/thread.py
index 723b74e..bf69d3e 100644
--- a/multiparser/thread.py
+++ b/multiparser/thread.py
@@ -124,8 +124,8 @@ def _prepare_parsed_data(
             for d in _data:
                 _row_metadata = _metadata.copy()
                 if _row_number is not None and d:
-                    _row_number += 1
                     _row_metadata["row_number"] = _row_number
+                    _row_number += 1
                 yield (_row_metadata, d)
             return
     elif (
diff --git a/tests/test_data_parsing.py b/tests/test_data_parsing.py
index 844df96..a95c4f2 100644
--- a/tests/test_data_parsing.py
+++ b/tests/test_data_parsing.py
@@ -174,7 +174,7 @@ def test_parse_delimited(fake_delimited_log, request, header) -> None:
 
     for _ in range(10):
         time.sleep(0.1)
-        _, _parsed_data = mp_parse.record_log(
+        _metadata, _parsed_data = mp_parse.record_log(
             input_file=_file,
             tracked_values=None,
             parser_func=record_with_delimiter,
@@ -306,7 +306,7 @@ def parse():
     with path.open("a") as stream:
         stream.write("value\n\n10\n# ignored\n20\n\n")
     parse()
-    assert [meta["row_number"] for _, meta in captured] == [1, 2]
+    assert [meta["row_number"] for _, meta in captured] == [0, 1]
     assert [data for data, _ in captured] == [{"value": 10}, {"value": 20}]
     assert metadata["__rows_read"] == 2
     assert "__read_bytes" in metadata
@@ -331,11 +331,11 @@ def parse():
     with path.open("a") as stream:
         stream.write("30\n\n40\n")
     parse()
-    assert [meta["row_number"] for _, meta in captured] == [1, 2, 3, 4]
+    assert [meta["row_number"] for _, meta in captured] == [0, 1, 2, 3]
     assert captured[-1][0] == {"value": 40}
     assert metadata["__rows_read"] == 4
     # Each callback gets independent metadata, not an overwritten shared dict.
-    assert captured[0][1]["row_number"] == 1
+    assert captured[0][1]["row_number"] == 0
 
     parse()  # No new content must not repeat any rows or lose state.
     assert len(captured) == 4
@@ -360,7 +360,7 @@ def test_direct_csv_parser_row_numbers(tmp_path, headers):
         for meta, data in _prepare_parsed_data(parsed)
         if data
     ]
-    assert rows == [(1, {"x": 1, "y": 2}), (2, {"x": 3, "y": 4})]
+    assert rows == [(0, {"x": 1, "y": 2}), (1, {"x": 3, "y": 4})]
 
 
 @pytest.mark.parsing
@@ -378,7 +378,7 @@ def parser(file_content, **kwargs):
     assert metadata["__read_bytes"] == path.stat().st_size
     rows = [(meta, row) for meta, row in _prepare_parsed_data((metadata, data)) if row]
     assert len(rows) == 1
-    assert rows[0][0]["row_number"] == 1
+    assert rows[0][0]["row_number"] == 0
     assert rows[0][0]["custom"] == "metadata"
 
     with path.open("a") as stream:
@@ -386,7 +386,7 @@ def parser(file_content, **kwargs):
     metadata, data = mp_parse.record_log(str(path), parser_func=parser, **metadata)
     rows = [(meta, row) for meta, row in _prepare_parsed_data((metadata, data)) if row]
     assert len(rows) == 1
-    assert rows[0][0]["row_number"] == 2
+    assert rows[0][0]["row_number"] == 1
     assert metadata["__rows_read"] == 2