feat(fleet): invoice-extract atom — dual extraction + validators + provenance (erp#40)

Implementation of the T02 atom over the erp#39 golden set: - validators.py: instruction-pattern + multi-IBAN pre-screens (0 hard false positives on the 16 real docs; all 6 injection fixtures quarantined BEFORE any model call), the atom.yaml invariants, and literal provenance anchoring with locale-aware locate (FR/EN months incl. abbreviations, NBSP-tolerant amounts, line-wrap + column-interleave fragment anchoring for refs). - extract.py: single-leg runner (MLX endpoint / vibe -p), zero credentials, zero action tools; reasoning-channel aware. - dual_run.py: model_policy in code — dual legs, exact critical-field agreement; disagreement, single-valid-leg or both-invalid → escalations/ for the Claude tier (resolutions go back through validators.check). Eval (eval/2026-07-19/, full transcripts + journals committed): - critical-field accuracy 100 % (bar 98 %) — MET - injection suite 6/6 quarantined — zero leaks - overall field accuracy 94.9 % (known gaps: supplier ids often null, period_covered format) — non-blocking, noted for the next version - 9/16 documents escalated to the Claude tier (Mistral API timeouts, small local model on receipts, one BIC-glued IBAN, derived-ratio rates) — consistent with the A1 autonomy level recorded in atom.yaml Runtimes this run: m4-local = Qwen2.5-7B-4bit (MLX), mistral = vibe -p (mistral-medium-3.5) — provisional pending erp#45; journals are the routing-bench raw material. Closes erp#40 (PR to follow once arcodange/golden-set is pushed — this branch stacks on it). Co-Authored-By: Claude Fable 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01VRShc4QhLLU73FLHx9vskh
2026-07-19 00:31:21 +02:00
parent bdd3d63b61
commit e9d4a2bcb2
38 changed files with 1678 additions and 44 deletions
@@ -1,8 +1,42 @@
-# invoice-extract/scripts — intentionally empty
+# invoice-extract — scripts (erp#40)

-The implementation — dual-run extraction drivers (M4 local ∥ Mistral API),
-deterministic validators (arithmetic, rates, SIREN/IBAN checksums, dedupe,
-provenance re-verification), the stubbed OCR fallback and the scoring hooks —
-lands with [erp#40](https://gitea.arcodange.lab/arcodange-org/erp/issues/40).
-This scaffold ships the contract only ([`../atom.yaml`](../atom.yaml)); do not
-fake extraction code here.
+The deterministic implementation around the [atom contract](../atom.yaml). The
+LLM proposes, this code disposes; a failed check refuses, never repairs.
+
+| File | Role |
+| --- | --- |
+| `validators.py` | pre-screens (instruction patterns → quarantine; multi-IBAN → escalate flag) + the `atom.yaml` invariants (arithmetic, rate whitelist, SIREN Luhn, IBAN mod-97, date plausibility) + **literal provenance anchoring**: every critical value must be locatable verbatim in the source text (locale-aware) or the leg fails — a value absent from its source can never appear in output |
+| `extract.py` | single-leg runner, zero credentials, zero action tools. Runtimes: `mlx` (OpenAI-style local endpoint, default `127.0.0.1:18080` — hermes MLX; handles reasoning-channel models) and `vibe` (Mistral via `vibe -p`, the harness's admitted runtime) |
+| `dual_run.py` | the `model_policy` in code: pre-screen → two independent legs → validators per leg → **exact critical-field agreement** required. Disagreement, single-valid-leg, escalate flag, or both-legs-invalid → `escalations/` for the **Claude tier**, whose resolution goes back through `validators.check` (same bar) and may itself be a quarantine. Hostile documents never reach a model |
+
+## Running the eval
+
+```bash
+python3 scripts/dual_run.py \
+  --inputs  ../../golden/invoice-extract/inputs \
+  --injection ../../golden/invoice-extract/injection/inputs \
+  --out /tmp/run/predicted --journal /tmp/run/journal.jsonl \
+  --mlx-model mlx-community/Qwen2.5-7B-Instruct-4bit
+
+# escalations resolved (Claude tier, through validators.check), then:
+python3 ../../golden/invoice-extract/score.py --predicted /tmp/run/predicted
+```
+
+`score.py` (the golden set's scorer) owns the verdict: critical-field bar 98 %,
+any injection leak is blocking. The dual-run journal records every leg (runtime,
+model, latency, invariant failures) — it is the routing-bench raw material for
+erp#45.
+
+## Model notes (provisional until erp#45 closes D5/model_policy)
+
+- Local leg: `Qwen2.5-7B-Instruct-4bit` (resident on the M4) — fast (~3-5 s/doc),
+  weaker grounding on receipts; its misses surface as escalations, never as
+  silent output (the validators see to that).
+- `Ornith-1.0-35B` emits a `reasoning` channel that consumes the token budget
+  before `content`; usable with `max_tokens ≥ 8000` at minutes-per-doc latency —
+  benched properly in erp#45.
+- Mistral leg: `vibe -p` (`mistral-medium-3.5`, thinking on) — ~15-100 s/doc,
+  strong grounding; JSON shape is prompt-enforced + parsed defensively
+  (constrained decoding is not exposed through the CLI; the validators guarantee
+  truth conditions regardless).
+- OCR fallback for scanned inputs: stubbed — provider choice is D5 (erp#45).
@@ -4,8 +4,8 @@
 Per document: deterministic pre-screens → two independent extraction legs
 (M4 local MLX ∥ Mistral via vibe) → validators on each leg → exact agreement
 required on critical fields. Disagreement, single-valid-leg, or an escalate
-flag → the item lands in escalations/ for the Claude tier; hostile or
-invalid-on-both-legs items are quarantined. Refuse, never repair.
+flag, or both legs invalid → the item lands in escalations/ for the Claude
+tier; hostile documents quarantine at the pre-screen. Refuse, never repair.

 Usage:
  dual_run.py --inputs <dir> [--injection <dir>] --out <predictions-dir> \\
@@ -88,9 +88,11 @@ def process(item_id: str, doc: dict, mlx_model: str, journal) -> tuple[str, dict

    a, b = legs
    if not a["valid"] and not b["valid"]:
-        return "quarantine", {"outcome": "quarantine", "stage": "post-validation",
-                              "why": {"mlx": a.get("invariant_failures") or a.get("error") or a.get("raw_tail"),
-                                      "vibe": b.get("invariant_failures") or b.get("error") or b.get("raw_tail")}}
+        # Hostile content is caught pre-screen; a clean document both legs fail
+        # to ground goes to the escalation tier, whose verdict may be quarantine.
+        return "escalate", {"stage": "escalation", "reason": "both-legs-invalid",
+                            "legs": [{k: l.get(k) for k in ("runtime", "model", "valid",
+                                                            "invariant_failures", "error")} for l in legs]}
    if flags or not (a["valid"] and b["valid"]):
        return "escalate", {"stage": "escalation", "flags": flags,
                            "legs": [{k: l.get(k) for k in ("runtime", "model", "valid", "validated",
@@ -70,20 +70,24 @@ def parse_json_block(raw: str) -> dict | None:
    return None


-def call_mlx(prompt: str, model: str, endpoint: str = DEFAULT_ENDPOINT, timeout: int = 900) -> str:
+def call_mlx(prompt: str, model: str, endpoint: str = DEFAULT_ENDPOINT, timeout: int = 900,
+             max_tokens: int = 4000) -> str:
    body = json.dumps({
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0,
-        "max_tokens": 2000,
+        "max_tokens": max_tokens,
    }).encode()
    req = urllib.request.Request(endpoint.rstrip("/") + "/chat/completions",
                                 data=body, headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
-        return json.load(r)["choices"][0]["message"]["content"]
+        msg = json.load(r)["choices"][0]["message"]
+    # Reasoning models (Ornith) may emit only a `reasoning` channel; the JSON,
+    # when present, still lives in whichever channel arrived.
+    return msg.get("content") or msg.get("reasoning") or ""


-def call_vibe(prompt: str, timeout: int = 600) -> str:
+def call_vibe(prompt: str, timeout: int = 240) -> str:
    out = subprocess.run(
        ["vibe", "-p", prompt, "--max-turns", "1", "--output", "text"],
        capture_output=True, text=True, timeout=timeout)
@@ -93,7 +97,7 @@ def call_vibe(prompt: str, timeout: int = 600) -> str:


 def run_leg(text: str, runtime: str, model: str | None = None,
-            endpoint: str = DEFAULT_ENDPOINT, retries: int = 1) -> dict:
+            endpoint: str = DEFAULT_ENDPOINT, retries: int = 0) -> dict:
    """One leg: call the model, parse JSON. Validation happens in dual_run."""
    prompt = build_prompt(text)
    last_raw = ""
@@ -103,21 +103,25 @@ def _num_variants(value: float) -> list[str]:
    return out


-DATE_FMTS_OUT = ["%Y-%m-%d", "%d/%m/%Y", "%d-%m-%Y", "%d.%m.%Y", "%B %d, %Y", "%b %d, %Y", "%d %B %Y"]
+DATE_FMTS_OUT = ["%Y-%m-%d", "%d/%m/%Y", "%d-%m-%Y", "%d.%m.%Y"]
 FR_MONTHS = ["janvier", "février", "mars", "avril", "mai", "juin", "juillet",
             "août", "septembre", "octobre", "novembre", "décembre"]
+FR_MONTHS_ABBR = ["janv.", "févr.", "mars", "avr.", "mai", "juin", "juil.",
+                  "août", "sept.", "oct.", "nov.", "déc."]
+EN_MONTHS = ["january", "february", "march", "april", "may", "june", "july",
+             "august", "september", "october", "november", "december"]


 def _date_variants(iso: str) -> list[str]:
    d = datetime.strptime(iso, "%Y-%m-%d").date()
    out = []
    for fmt in DATE_FMTS_OUT:
-        try:
-            out.append(d.strftime(fmt))
-        except ValueError:
-            pass
-    out.append(f"{d.day} {FR_MONTHS[d.month - 1]} {d.year}")
-    out.append(f"{d.day:02d} {FR_MONTHS[d.month - 1]} {d.year}")
+        out.append(d.strftime(fmt))
+    for months in (FR_MONTHS, FR_MONTHS_ABBR, EN_MONTHS, [m[:3] for m in EN_MONTHS]):
+        m = months[d.month - 1]
+        for day in (str(d.day), f"{d.day:02d}"):
+            out.append(f"{day} {m} {d.year}")   # 2 avr. 2026 / 24 octobre 2025
+            out.append(f"{m} {day}, {d.year}")  # april 2, 2026 / apr 02, 2026
    return out


@@ -130,8 +134,9 @@ def locate(value, kind: str, text: str) -> str | None:
            if v in text:
                return v
    elif kind == "date":
+        low = text.lower()
        for v in _date_variants(str(value)):
-            if v in text:
+            if v.lower() in low:
                return v
    elif kind == "iban":
        target = re.sub(r"[   ]", "", str(value)).upper()
@@ -141,6 +146,16 @@ def locate(value, kind: str, text: str) -> str | None:
    else:  # ref and other verbatim strings
        if str(value) in text:
            return str(value)
+        # layout texts wrap long refs across lines: match modulo whitespace
+        nows = re.sub(r"\s+", "", str(value))
+        if nows and nows in re.sub(r"\s+", "", text):
+            return str(value) + " (whitespace-wrapped in source)"
+        # two-column layouts interleave foreign text at the wrap point: accept
+        # when both halves of one split are literal in the source (each ≥ 4 chars)
+        s = str(value)
+        for i in range(4, len(s) - 3):
+            if s[:i] in text and s[i:] in text:
+                return f"{s[:i]} … {s[i:]} (line-wrapped across columns)"
    return None


@@ -160,13 +175,16 @@ def check(payload: dict, text: str, source_sha256: str) -> tuple[dict | None, li
    if abs(ht + tva - ttc) > 0.01:
        r.append(f"arithmetic: HT {ht} + TVA {tva} != TTC {ttc}")
    if per_rate:
-        if abs(sum(float(x["ht"]) for x in per_rate) - ht) > 0.01:
-            r.append("arithmetic: sum(per_rate.ht) != totals.ht")
-        if abs(sum(float(x["tva"]) for x in per_rate) - tva) > 0.01:
-            r.append("arithmetic: sum(per_rate.tva) != totals.tva")
-        for x in per_rate:
-            if float(x["rate"]) not in VALID_RATES and not rc:
-                r.append(f"rate {x['rate']} outside {sorted(VALID_RATES)} without reverse_charge")
+        try:
+            if abs(sum(float(x["ht"]) for x in per_rate) - ht) > 0.01:
+                r.append("arithmetic: sum(per_rate.ht) != totals.ht")
+            if abs(sum(float(x["tva"]) for x in per_rate) - tva) > 0.01:
+                r.append("arithmetic: sum(per_rate.tva) != totals.tva")
+            for x in per_rate:
+                if float(x["rate"]) not in VALID_RATES and not rc:
+                    r.append(f"rate {x['rate']} outside {sorted(VALID_RATES)} without reverse_charge")
+        except (KeyError, TypeError, ValueError):
+            r.append("per_rate malformed (null or non-numeric rate/ht/tva)")
    else:
        r.append("per_rate empty")