fix(csv-import): thread the reserved column through every header reader

9182e0f #3 excluded original_row_order from the totality check by pre-stripping
the header at the call site (`spec_header`). Three other surfaces still read the
raw or stripped header inconsistently, breaking exactly in the path #3 enables.
Thread the reserved column through _normalize_schema (and _suggest_schema_dsl)
as a `reserved` param instead, so the coverage check, the diagnostics, and
positional binding all read one consistent header:

- Suggested-schema hint (#1): _suggest_schema_dsl drops reserved columns, so the
  #143 recovery suggestion is paste-ready. It no longer emits a cell for
  original_row_order, which the totality check excludes — pasting the old
  suggestion back raised 'positional schema has 3 columns but the header has 2'.
- Diagnostics (#2): error messages report the speccable header plus a note about
  the reserved column, instead of the reserved-stripped subset. A 3-column file
  is no longer described as 'has 2 [a, b]', and a by-name schema that names the
  reserved column gets a targeted message, not a misleading 'not in the header'.
- Positional binding (#3): a canonical original_row_order that is not the
  trailing column now raises (steering to a by-name schema) rather than silently
  shifting later positional cells onto the wrong data column.

Also rejects a spec that renames a DATA column ONTO original_row_order up front,
instead of leaking a raw polars DuplicateError at sink.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: paddymul

src/tallyman_xorq/io.py

@@ -207,19 +207,39 @@ def _spec_pairs(spec):
     return list(spec.items())
 
 
-def _normalize_schema(spec, header: list[str]) -> dict:
+def _normalize_schema(spec, header: list[str], *, reserved: tuple[str, ...] = ()) -> dict:
     """Resolve a schema spec against the CSV *header* into a polars parse plan.
 
+    *header* is the CSV's full header. *reserved* names tallyman-managed columns
+    (currently ``original_row_order``) that the caller must not spec: they are
+    excluded from the columns the spec has to cover, but kept in the diagnostics
+    so an error message matches the file the user is looking at rather than a
+    silently reserved-stripped subset. Threading ``reserved`` in — rather than
+    pre-stripping the header at the call site — keeps the totality check, the
+    diagnostics, and positional binding all reading one consistent header.
+
     Returns ``{overrides, rename, out_names, all_pinned}``:
       - ``overrides``  — ``{header_name: polars_dtype}`` for pinned columns, keyed by the
         name polars reads (positional renames are applied *after* the read).
       - ``rename``     — ``{header_name: output_name}`` where they differ (positional only).
       - ``out_names``  — output column names in column order (the final ``select``).
       - ``all_pinned`` — True if no column is inferred (then ``infer_schema_length=0``).
 
-    Raises the actionable ValueError on a by-name miss, a non-total spec, or an
-    over-long positional spec — these are the #143 error-contract surfaces.
+    Raises the actionable ValueError on a by-name miss, a non-total spec, an
+    over-long positional spec, a by-name spec that names a reserved column, or a
+    positional spec whose reserved column is not the trailing column — these are
+    the #143 error-contract surfaces.
     """
+    reserved = tuple(r for r in reserved if r in header)
+    speccable = [h for h in header if h not in reserved]
+    reserved_hint = ""
+    if reserved:
+        names = ", ".join(repr(r) for r in reserved)
+        reserved_hint = (
+            f" (the file also has tallyman's reserved column {names}, which tallyman "
+            "manages automatically — leave it out of the schema.)"
+        )
+
     overrides: dict = {}
     rename: dict = {}
     inferred = False
@@ -237,20 +257,31 @@ def _normalize_schema(spec, header: list[str]) -> dict:
             cells = cells[:-1]
         if any(n in (_REST, _BEGIN) for n, _ in cells):
             raise ValueError("tallyman_read_csv: '&rest' must be the last cell and '&begin' the first.")
-        if len(cells) > len(header):
+        # Positional cells bind by physical column position. Excluding a reserved column
+        # that is not the trailing column would silently shift every later cell onto the
+        # wrong data column, so require the reserved column(s) to be a trailing suffix
+        # (header[:len(speccable)] == speccable); otherwise steer to a by-name schema.
+        if reserved and header[: len(speccable)] != speccable:
+            raise ValueError(
+                f"tallyman_read_csv: a positional schema cannot bind {list(header)} — its reserved "
+                f"column {list(reserved)} is not the last column, and positional cells bind by physical "
+                "column position, so excluding it would silently rebind later cells onto the wrong data "
+                "column. Use a by-name schema (a dict or ibis schema keyed on the header names) instead."
+            )
+        if len(cells) > len(speccable):
             raise ValueError(
                 f"tallyman_read_csv: positional schema has {len(cells)} columns but the CSV header has "
-                f"{len(header)} {list(header)}."
+                f"{len(speccable)} {list(speccable)}.{reserved_hint}"
             )
-        if rest_dtype is None and len(cells) != len(header):
-            uncovered = list(header[len(cells):])
+        if rest_dtype is None and len(cells) != len(speccable):
+            uncovered = list(speccable[len(cells):])
             raise ValueError(
                 f"tallyman_read_csv: schema is not total — it leaves column(s) {uncovered} unspecified. "
                 'Name every column, or close the spec with ("&rest", "infer").'
             )
         out_names = []
         for i, (out_name, dt) in enumerate(cells):
-            orig = header[i]
+            orig = speccable[i]
             out_names.append(out_name)
             if out_name != orig:
                 rename[orig] = out_name
@@ -259,7 +290,7 @@ def _normalize_schema(spec, header: list[str]) -> dict:
                 inferred = True
             else:
                 overrides[orig] = pl_dt
-        for orig in header[len(cells):]:  # the &rest tail keeps its header name
+        for orig in speccable[len(cells):]:  # the &rest tail keeps its header name
             out_names.append(orig)
             pl_dt = _resolve_spec_dtype(rest_dtype)
             if pl_dt is None:
@@ -278,14 +309,20 @@ def _normalize_schema(spec, header: list[str]) -> dict:
             rest_dtype = dt
             continue
         named[str(name)] = dt
-    missing = [n for n in named if n not in header]
+    named_reserved = [n for n in named if n in reserved]
+    if named_reserved:
+        raise ValueError(
+            f"tallyman_read_csv: schema names tallyman's reserved column {sorted(named_reserved)}, which "
+            "tallyman manages automatically — remove it from the schema (it is added and ordered for you)."
+        )
+    missing = [n for n in named if n not in speccable]
     if missing:
         raise ValueError(
-            f"tallyman_read_csv: schema column(s) {sorted(missing)} are not in the CSV header {list(header)}. "
+            f"tallyman_read_csv: schema column(s) {sorted(missing)} are not in the CSV header {list(speccable)}. "
             "Bind by name only when the names match the header; to rename by position pass a tuple-of-tuples "
-            'schema, e.g. schema=(("Date", "date"), ("&rest", "infer")).'
+            'schema, e.g. schema=(("Date", "date"), ("&rest", "infer")).' + reserved_hint
         )
-    uncovered = [h for h in header if h not in named]
+    uncovered = [h for h in speccable if h not in named]
     if rest_dtype is None and uncovered:
         raise ValueError(
             f"tallyman_read_csv: schema is not total — it leaves column(s) {uncovered} unspecified. "
@@ -303,7 +340,7 @@ def _normalize_schema(spec, header: list[str]) -> dict:
             inferred = True
         else:
             overrides[name] = pl_dt
-    return {"overrides": overrides, "rename": {}, "out_names": list(header), "all_pinned": not inferred}
+    return {"overrides": overrides, "rename": {}, "out_names": list(speccable), "all_pinned": not inferred}
 
 
 _ESCALATED_INFER = 10_000  # the middle rung of the #143 infer ladder; whole-file (None) is the terminal
@@ -335,12 +372,18 @@ def _polars_to_ibis(dt) -> str:
     return "string"  # safe fallback — an unmapped column still reads as string
 
 
-def _suggest_schema_dsl(src: Path, scan_kwargs: dict) -> str:
-    """Whole-file-infer *src* and render it as the paste-ready tuple-of-tuples DSL."""
+def _suggest_schema_dsl(src: Path, scan_kwargs: dict, reserved: tuple[str, ...] = ()) -> str:
+    """Whole-file-infer *src* and render it as the paste-ready tuple-of-tuples DSL.
+
+    *reserved* columns (tallyman-managed, e.g. ``original_row_order``) are dropped:
+    they must not appear in the suggestion because the caller cannot spec them —
+    a suggestion carrying one is un-pasteable (the totality check excludes the
+    reserved column, so pasting it back over-counts the columns and raises).
+    """
     import polars as pl
 
     inferred = pl.scan_csv(str(src), infer_schema_length=None, **scan_kwargs).collect_schema()
-    pairs = tuple((name, _polars_to_ibis(dt)) for name, dt in inferred.items())
+    pairs = tuple((name, _polars_to_ibis(dt)) for name, dt in inferred.items() if name not in reserved)
     return repr(pairs)
 
 
@@ -398,18 +441,28 @@ def _materialize_ordered(src: Path, schema, scan_kwargs: dict, tmp_path: Path) -
     # check; infer_schema_length=0 reads just the header line, no type sampling.
     header = list(pl.scan_csv(str(src), infer_schema_length=0, **scan_kwargs).collect_schema().names())
     has_row_order = "original_row_order" in header
+    # original_row_order is tallyman's reserved index — _ordered re-appends it, and
+    # _validate_existing_row_order (below) vets a pre-existing one. It is not the caller's
+    # to spec, so it is threaded as `reserved` through every header reader (the schema
+    # plan, the diagnostics, and the suggested-schema hint) rather than stripped ad hoc.
+    reserved = ("original_row_order",) if has_row_order else ()
     if has_row_order:
         _validate_existing_row_order(src, scan_kwargs)  # reuse it, or raise — never with_row_index over it
 
     plan = None
     explicit_only = False
     if schema is not None:
-        # original_row_order is tallyman's reserved index — _ordered re-appends it, and
-        # _validate_existing_row_order already vetted a pre-existing one. It is not the
-        # caller's to spec, so exclude it from the header the spec must totally cover;
-        # otherwise a complete data-column schema is spuriously rejected as "not total".
-        spec_header = [h for h in header if h != "original_row_order"]
-        plan = _normalize_schema(schema, spec_header)
+        plan = _normalize_schema(schema, header, reserved=reserved)
+        # A spec may not PRODUCE the reserved name either: renaming a data column onto
+        # 'original_row_order' collides with the index _ordered re-appends (a raw polars
+        # DuplicateError at sink). Reject the reserved output name up front with the same
+        # guidance as the pre-existing-column clash.
+        if "original_row_order" in plan["out_names"]:
+            raise ValueError(
+                "tallyman_read_csv: 'original_row_order' is reserved for tallyman's canonical "
+                "row index and cannot be a schema output column name. Rename that column to "
+                "something other than 'original_row_order' in the schema."
+            )
         explicit_only = plan["all_pinned"]
 
     def _ordered(infer_len):
@@ -440,7 +493,7 @@ def _ordered(infer_len):
             last_exc = exc
     raise ValueError(
         f"tallyman_read_csv: the schema does not parse {src.name}: {last_exc}. "
-        f"Suggested schema (whole-file inference): schema={_suggest_schema_dsl(src, scan_kwargs)}"
+        f"Suggested schema (whole-file inference): schema={_suggest_schema_dsl(src, scan_kwargs, reserved)}"
     )