fill: fix label-collision silent overwrite + pikepdf handle leak

src/maine_forms_engine/fill/field_split.py

@@ -120,7 +120,10 @@ def split_to_copy(src_pdf: pathlib.Path, dst_pdf: pathlib.Path,
         return 0
     import pikepdf
 
-    pdf = pikepdf.open(str(src_pdf))
-    n = split(pdf, specs)
-    pdf.save(str(dst_pdf))
+    # Context manager so the file handle is released even if split()/save()
+    # raises — this runs per fill inside the long-lived MCP server every
+    # sibling repo hosts, where leaked handles accumulate.
+    with pikepdf.open(str(src_pdf)) as pdf:
+        n = split(pdf, specs)
+        pdf.save(str(dst_pdf))
     return n

src/maine_forms_engine/fill/fill_via_mapping.py

@@ -345,10 +345,51 @@ def fill_via_mapping(form_id: str, facts: dict, out_dir: pathlib.Path,
     fid_to_widgets: dict[str, list[str]] = {}
     for f in res["_schema"]["fields"]:
         fid_to_widgets.setdefault(f["field_id"], []).append(f["label"])
+    # Build the label-keyed fill dict. Because fill_form is keyed by widget
+    # LABEL, two *different* field_ids that resolve to the SAME widget label
+    # collide: the later write silently overwrites the earlier one and only
+    # one value reaches the PDF. That is distinct from the shared-AcroForm-
+    # field case (ONE field_id, many appearances) that field_split.py handles
+    # by detaching an appearance; here it is genuinely conflicting values for
+    # one box, so we can't split it — we surface it. We also make `resolved`
+    # truthful: it must reflect widgets actually written, not the resolved
+    # field_id count (which double-counts field_ids the collision dropped).
     field_data: dict[str, str] = {}
+    # label -> the field_id that first claimed it (for conflict reporting).
+    label_owner: dict[str, str] = {}
+    label_conflicts: list[dict] = []
+    # field_ids that resolved to a value but had ALL their widget labels
+    # overwritten by a later, differing field_id (so nothing they resolved
+    # actually reached the PDF).
+    shadowed_fids: set[str] = set()
     for fid, v in fitted.items():
         for label in fid_to_widgets.get(fid, []):
+            if label in field_data and field_data[label] != v:
+                # Different field_id, different value, same widget label:
+                # a silent overwrite. Record it. Last-write-wins is kept so
+                # the PDF byte-output is unchanged from prior behavior — the
+                # fix is to stop lying about it, not to reshuffle who wins.
+                prev_fid = label_owner.get(label)
+                label_conflicts.append({
+                    "label": label,
+                    "kept_field_id": fid,
+                    "kept_value": v,
+                    "dropped_field_id": prev_fid,
+                    "dropped_value": field_data[label],
+                })
+                if prev_fid is not None:
+                    shadowed_fids.add(prev_fid)
             field_data[label] = v
+            label_owner[label] = fid
+    # A field_id is only truly shadowed if NONE of its labels survived. Re-add
+    # any that still own at least one label (e.g. a fid backing two widgets
+    # where only one collided).
+    surviving_fids = set(label_owner.values())
+    shadowed_fids -= surviving_fids
+    # Truthful resolution count: resolved field_ids minus those whose written
+    # value was entirely dropped by a collision. With no collisions this is
+    # unchanged (== res["resolved"]), so existing consumers see no drift.
+    resolved_written = res["resolved"] - len(shadowed_fids)
     out_dir.mkdir(parents=True, exist_ok=True)
     # Split any shared AcroForm fields (forms/<ID>/field_splits.json) on a
     # working copy first, so a value mapped to one appearance no longer fans
@@ -411,12 +452,16 @@ def fill_via_mapping(form_id: str, facts: dict, out_dir: pathlib.Path,
                "computation_notes"):
         if _k in res:
             _extra[_k] = res[_k]
+    if label_conflicts:
+        # Additive diagnostic: two field_ids resolved to one widget label; the
+        # loser's value never reached the PDF. Present on both result styles.
+        _extra["label_conflicts"] = label_conflicts
     if result_style == "tax":
         return {
             **_extra,
             "form_id": form_id, "ok": True, "status": res["status"],
             "out_pdf": str(out_pdf),
-            "mapped_keys": res["mapped_keys"], "resolved": res["resolved"],
+            "mapped_keys": res["mapped_keys"], "resolved": resolved_written,
             # canonical keys that resolved to nothing in the case object
             "unresolved": [list(u) for u in res["unresolved"]],
             # widgets actually written, counted by the filler (not the request)
@@ -429,8 +474,8 @@ def fill_via_mapping(form_id: str, facts: dict, out_dir: pathlib.Path,
         }
     out = {**_extra,
            "form_id": form_id, "ok": True, "out_pdf": str(out_pdf),
-           "mapped_keys": res["mapped_keys"], "resolved": res["resolved"],
-           "coverage": (round(res["resolved"] / res["mapped_keys"], 3)
+           "mapped_keys": res["mapped_keys"], "resolved": resolved_written,
+           "coverage": (round(resolved_written / res["mapped_keys"], 3)
                         if res["mapped_keys"] else 0.0),
            "unresolved": [{"field_id": fid, "key": key}
                           for fid, key in res["unresolved"]],
@@ -439,7 +484,7 @@ def fill_via_mapping(form_id: str, facts: dict, out_dir: pathlib.Path,
                             "detail": blank_detail}}
     if split_skipped:
         out["split_step_skipped"] = split_skipped
-    if res["resolved"] == 0 and res["mapped_keys"]:
+    if resolved_written == 0 and res["mapped_keys"]:
         # A zero-resolved fill is a blank PDF — almost always a fact-object
         # shape problem (engine-shape case passed to the canonical-mapping
         # path). Surface it as a failure instead of a silent near-blank.

tests/test_field_split.py

@@ -78,3 +78,33 @@ def test_no_specs_means_no_copy(tmp_path):
     assert specs_for("TEST-N", root) == []
     assert split_to_copy(tmp_path / "in.pdf", tmp_path / "out.pdf",
                          "TEST-N", root) == 0
+
+
+def test_split_to_copy_closes_the_pdf_handle(shared_field_tree, tmp_path,
+                                              monkeypatch):
+    """Regression (audit 2026-07-06): split_to_copy used to open the source
+    PDF without a context manager, leaking one handle per fill inside the
+    long-lived MCP server. Assert the opened Pdf is closed before return."""
+    import maine_forms_engine.fill.field_split as fs
+
+    opened = []
+    real_open = pikepdf.open
+
+    def tracking_open(*a, **kw):
+        pdf = real_open(*a, **kw)
+        opened.append(pdf)
+        return pdf
+
+    monkeypatch.setattr(pikepdf, "open", tracking_open)
+    # field_split imports pikepdf lazily inside the function, so the patched
+    # module attribute is what it resolves.
+    src = shared_field_tree / "TEST-S" / "TEST-S.pdf"
+    dst = tmp_path / "TEST-S.split.pdf"
+    n = fs.split_to_copy(src, dst, "TEST-S", shared_field_tree)
+    assert n == 1
+    assert opened, "split_to_copy did not open the source PDF"
+    # every Pdf opened by split_to_copy must be closed on return. pikepdf
+    # marks a closed handle by resetting .filename to "closed input source".
+    for pdf in opened:
+        assert pdf.filename == "closed input source", (
+            "split_to_copy leaked an open pikepdf handle")

tests/test_fill_via_mapping.py

@@ -1,6 +1,7 @@
 """End-to-end mapping fill over a synthetic consumer-repo tree (modeled on
 maine-court-forms tests/test_fill_smoke.py, with the unshipped official
 blanks replaced by an in-test fixture form)."""
+import hashlib
 import json
 import warnings
 
@@ -11,7 +12,7 @@
     fill_via_mapping, resolve_mapping)
 from maine_forms_engine.fill.verify_fill import verify_fill
 
-from conftest import CASE
+from conftest import CASE, synthetic_form
 
 
 def test_resolve_mapping_coverage(form_tree):
@@ -158,3 +159,87 @@ def test_tax_result_style_diagnostics(form_tree, tmp_path):
     # + the __wrap_cache_ entry the filler adds for the group).
     assert res["fields_written"] == 4
     assert res["unresolved"] == []
+
+
+# --- label-collision silent-overwrite regression (audit 2026-07-06) ---------
+#
+# fill_form is keyed by widget LABEL. Two DIFFERENT field_ids that resolve to
+# the SAME widget label collide: the later write silently overwrites the
+# earlier one, only one value reaches the PDF, yet `resolved` (and therefore
+# `coverage`) used to count BOTH — over-reporting coverage on a blank box.
+
+def _collision_tree(tmp_path, *, distinct_values=True):
+    """A forms/TEST-C tree whose schema gives two text field_ids the SAME
+    widget label ('name_field'), each mapped to a different case key."""
+    root = tmp_path / "collision_repo"
+    fdir = root / "forms" / "TEST-C"
+    (fdir / "examples").mkdir(parents=True)
+    blank = fdir / "TEST-C.pdf"
+    synthetic_form(blank)
+    schema = {"form_id": "TEST-C", "fields": [
+        {"field_id": "name_field", "label": "name_field",
+         "rect": [72, 100, 400, 120]},
+        # a SECOND, distinct field_id sharing name_field's widget label
+        {"field_id": "addr_field", "label": "name_field",
+         "rect": [72, 140, 400, 160]},
+    ]}
+    mapping = {"form_id": "TEST-C", "status": "verified", "map": {
+        "name_field": "parties.plaintiff.full_name",
+        "addr_field": "parties.plaintiff.address",
+    }}
+    addr = "1 Main St" if distinct_values else "Jane Q. Doe"
+    case = {"parties": {"plaintiff": {"full_name": "Jane Q. Doe",
+                                      "address": addr}}}
+    (fdir / "schema.json").write_text(json.dumps(schema))
+    (fdir / "mapping.json").write_text(json.dumps(mapping))
+    (fdir / "examples" / "sample_case.json").write_text(json.dumps(case))
+    data = blank.read_bytes()
+    (root / "catalog").mkdir()
+    (root / "catalog" / "pdf_manifest.json").write_text(json.dumps({
+        "count": 1, "forms": {"TEST-C": {
+            "url": "https://example.test/TEST-C",
+            "sha256": hashlib.sha256(data).hexdigest(),
+            "bytes": len(data)}}}))
+    return root, case
+
+
+def test_label_collision_makes_resolved_truthful(tmp_path):
+    root, case = _collision_tree(tmp_path, distinct_values=True)
+    res = fill_via_mapping("TEST-C", case, tmp_path / "out",
+                           forms_root=root / "forms")
+    assert res["ok"], res
+    # Two keys resolved from the case, but only ONE distinct value could land
+    # on the shared widget label — `resolved` must reflect what was written.
+    assert res["mapped_keys"] == 2
+    assert res["resolved"] == 1
+    assert res["coverage"] == 0.5
+    # ...and the dropped value is surfaced, not swallowed.
+    conflicts = res["label_conflicts"]
+    assert len(conflicts) == 1
+    c = conflicts[0]
+    assert c["label"] == "name_field"
+    assert {c["kept_field_id"], c["dropped_field_id"]} == {
+        "name_field", "addr_field"}
+    assert {c["kept_value"], c["dropped_value"]} == {"Jane Q. Doe", "1 Main St"}
+
+
+def test_label_collision_same_value_is_not_a_conflict(tmp_path):
+    # Two field_ids resolving to the SAME value on one label is a harmless
+    # idempotent write, not a data-loss collision — no conflict, no undercount.
+    root, case = _collision_tree(tmp_path, distinct_values=False)
+    res = fill_via_mapping("TEST-C", case, tmp_path / "out",
+                           forms_root=root / "forms")
+    assert res["ok"], res
+    assert "label_conflicts" not in res
+    assert res["resolved"] == 2
+
+
+def test_no_collision_leaves_resolved_and_result_shape_unchanged(form_tree,
+                                                                 tmp_path):
+    # Backward-compat guard for the four consumer repos that pin by tag: with
+    # no colliding labels, `resolved` is unchanged and no additive key appears.
+    res = fill_via_mapping("TEST-1", CASE, tmp_path / "out",
+                           forms_root=form_tree / "forms")
+    assert res["resolved"] == 4
+    assert res["coverage"] == 1.0
+    assert "label_conflicts" not in res

tests/test_form_filler.py

@@ -31,6 +31,20 @@ def test_word_wider_than_widget_skips_ahead(self):
         self.assertEqual(lines[1], "longword")
         self.assertEqual(rem, "ok")
 
+    @pytest.mark.xfail(reason="audit 2026-07-06 follow-up: a wide word that "
+                       "jumps ahead leaves the skipped widget blank; a later "
+                       "narrow word could fill it, but packing it there would "
+                       "reorder text out of top-to-bottom reading order "
+                       "(see PR discussion). Deferred as a design decision.",
+                       strict=True)
+    def test_skipped_widget_backfilled_by_later_word(self):
+        # 'bbbbbbbb' (8) skips widgets 0 and 1 (cap 5) to land in widget 2;
+        # 'cc' (2) fits the skipped widget 1. The gap currently stays blank:
+        # actual == (['aaaa', '', 'bbbbbbbb cc'], '').
+        lines, rem = _wrap_across_widgets("aaaa bbbbbbbb cc", [5, 5, 20])
+        self.assertNotEqual(lines[1], "")
+        self.assertEqual(rem, "")
+
 
 @pytest.fixture
 def out(tmp_path):