Document-to-JSON Pipeline

# INVOICE

**Invoice ID:** INV-2026-0042
**Vendor:** Northwind Traders, Inc.
**Issue Date:** 2026-05-01
**Due Date:** 2026-05-31
**Currency:** USD

Bill To: Caliber Demo Co., 500 Market St, Seattle WA

| # | Description                       | Quantity | Unit Price | Amount  |
|---|-----------------------------------|----------|------------|---------|
| 1 | Standard Widget (part WGT-100)    | 10       | 12.50      | 125.00  |
| 2 | Premium Widget (part WGT-200)     | 5        | 30.00      | 150.00  |
| 3 | On-site installation (per hour)   | 4        | 95.00      | 380.00  |

Subtotal: 655.00
Tax (8%): 52.40
**Total: 707.40 USD**

Payment terms: Net 30. Please reference the invoice ID on remittance.

<!--
STAND-IN for the GOLDEN case (dataset rows D01/D02).
Every required field of schema/extracted-fields.schema.json is present and
verifiable: invoice_id, vendor, line_items[] (3 rows), total (707.40),
currency (USD), issue_date (2026-05-01). Expected validation_status: pass.
Convert to a real .docx (D01) and .xlsx (D02) per sources/README.md before the
live extract path; for a dry run you can extract this .md directly.
-->

field,value
invoice_id,INV-2026-0042
vendor,"Northwind Traders, Inc."
issue_date,2026-05-01
due_date,2026-05-31
currency,USD
line_item,"Standard Widget (part WGT-100)|qty=10|unit_price=12.50|amount=125.00"
line_item,"Premium Widget (part WGT-200)|qty=5|unit_price=30.00|amount=150.00"
line_item,"On-site installation (per hour)|qty=4|unit_price=95.00|amount=380.00"
subtotal,655.00
tax,52.40
total,707.40

# INVOICE

**Invoice ID:** INV-2026-0099
**Vendor:** Contoso Office Supplies
**Issue Date:** 2026-05-10
**Currency:** USD

Bill To: Caliber Demo Co., 500 Market St, Seattle WA

| # | Description                  | Quantity | Unit Price | Amount  |
|---|------------------------------|----------|------------|---------|
| 1 | Managed print service (May)  | 1        | 300.00     | 300.00  |
| 2 | Toner cartridge (black)      | 2        | 45.00      | 90.00   |

Notes: This is a draft invoice. The grand total has NOT yet been finalized and
is intentionally omitted from this document — do not compute or infer it.

<!--
STAND-IN for the EDGE / missing-field case (dataset row D03).
Required fields invoice_id, vendor, line_items[], currency, issue_date are
present and verifiable, but `total` is deliberately absent. The doc-structurer
prompt must NOT invent a total; it should list "total" in missing_fields, and
validate_document_json must then return validation_status: partial with
missing_fields == ["total"]. Convert to a real .docx per sources/README.md, or
extract this .md directly for a dry run.
-->

# Source documents for SCN-04 — stand-ins + how to make real Office files

The extraction pipeline runs on **real binary Office files** (`.docx` / `.pptx`
/ `.xlsx`). Those are ZIP containers and **cannot be authored as text**, so this
folder ships **stand-in source content** you *can* read and edit:

| Stand-in file | Represents | Dataset row(s) | Expected outcome |
| --- | --- | --- | --- |
| [`invoice-clean.md`](invoice-clean.md) | clean invoice (all fields) | D01 (`.docx`) | `validation_status: pass` |
| [`invoice-clean.csv`](invoice-clean.csv) | same invoice as a sheet | D02 (`.xlsx`) | `validation_status: pass` |
| [`invoice-partial.md`](invoice-partial.md) | invoice missing `total` | D03 (`.docx`) | `partial`, `missing_fields:["total"]` |

The dataset ([`../extraction-cases.jsonl`](../extraction-cases.jsonl)) also
references three files that are **deliberately not provided as content** because
they only exist to exercise error paths — create them with the one-liners below:

- `invoice-no-vendor.docx` (D04) — copy `invoice-clean.md`, delete the
  **Vendor** line, save as `.docx`. Expect `partial`, `missing_fields:["vendor"]`.
- `invoice-legacy.doc` (D05) — a **legacy** Word doc (the unsupported-format
  negative case). See "Making the unsupported `.doc`" below.
- `invoice-corrupt.docx` (D06) — a deliberately broken `.docx`. See "Making the
  corrupt file" below.

## Dry run (no conversion) — fastest

`caliber.workflows.ingestion_tools:extract_document` reads `.md`, `.csv`, and
`.txt` as text directly. To smoke-test the **prompt + validator** wiring without
producing Office binaries, upload the `.md`/`.csv` stand-ins to the `doc-intake`
bucket and point the workflow at them. The text the structurer sees is
equivalent; only the `format` field of the extract output differs
(`markdown`/`text` instead of `docx`/`xlsx`). The golden/partial assertions in
the dataset still hold.

> Note: a dry run does **not** cover the negative cases (D05/D06) — those need a
> real legacy `.doc` and a real corrupt `.docx` (see below).

## Make the real Office files

### Option A — LibreOffice headless (no code; converts in place)

```sh
cd cookbooks/04-document-extraction-structuring-lab/assets/dataset/sources
# .md -> .docx  (D01, and the basis for D04 after you remove the Vendor line)
soffice --headless --convert-to docx invoice-clean.md
soffice --headless --convert-to docx invoice-partial.md
# .csv -> .xlsx (D02)
soffice --headless --convert-to xlsx invoice-clean.csv
```

### Option B — Python (`python-docx` + `openpyxl`, from the `caliber-suite[ingest]` extra)

```python
# invoice-clean.docx (D01)
from docx import Document
doc = Document()
for line in open("invoice-clean.md", encoding="utf-8"):
    line = line.rstrip("\n")
    if line.strip().startswith("<!--"):
        break  # stop at the trailing HTML comment
    doc.add_paragraph(line)
doc.save("invoice-clean.docx")

# invoice-clean.xlsx (D02)
import csv
from openpyxl import Workbook
wb = Workbook(); ws = wb.active; ws.title = "Invoice"
with open("invoice-clean.csv", newline="", encoding="utf-8") as fh:
    for row in csv.reader(fh):
        ws.append(row)
wb.save("invoice-clean.xlsx")
```

Then land each produced file on the platform host at the **exact absolute
local path** named in the dataset's `doc_path` (e.g. the
`.../assets/dataset/sources/invoice-clean.docx` placeholder). You can also upload
the files to the `doc-intake` bucket (`Object Store → doc-intake → Upload`) for
preview, but the extractor reads them by local path, not by bucket key.

## Making the unsupported `.doc` (negative case D05)

The extractor tool does **not** raise a typed error for a `.doc` — it would read
the binary as best-effort UTF-8 (garbage). The **readable** `kind:"unsupported"`
diagnostic is produced by the **Object Store extract endpoint**, which is the
surface this case targets. Produce a genuine legacy binary `.doc` and confirm
the endpoint rejects it:

```sh
# Real legacy .doc (binary Word 97-2003), NOT a renamed .docx:
soffice --headless --convert-to doc invoice-clean.md   # -> invoice-clean.doc
mv invoice-clean.doc invoice-legacy.doc
```

Upload `invoice-legacy.doc`, then in the UI open it and click **Extract** (or
call `GET /object-store/buckets/doc-intake/object/extract?key=invoice-legacy.doc`).
Expect a readable `kind:"unsupported"` error naming the format — that is the
`unsupported_format_returns_readable_error` rule check, and the workflow run
should surface it cleanly (status `fail`) rather than crash.

> Do **not** simply rename a `.docx` to `.doc`; that produces a Zip-with-a-`.doc`
> name, not a real legacy OLE document, and won't represent the case faithfully.

## Making the corrupt file (negative case D06)

Truncate a valid `.docx` so the parser fails mid-read:

```sh
# after producing invoice-clean.docx above:
head -c 2048 invoice-clean.docx > invoice-corrupt.docx   # truncated ZIP -> parse error
```

Upload `invoice-corrupt.docx`. The `extract_document` tool raises
`IngestionError("failed to extract docx ...")`; the workflow run ends `fail`
with that message attributed to the **extract** node (not the validate node) —
which is exactly the extraction-vs-validation separation the gate asks for.

{
  "name": "extract_document",
  "version": "1",
  "module_path": "caliber.workflows.ingestion_tools",
  "callable_name": "extract_document",
  "side_effect_level": "read",
  "allow_in_preview": true,
  "input_schema": {
    "type": "object",
    "required": ["ref"],
    "properties": {
      "ref": {
        "type": "string",
        "description": "Absolute LOCAL filesystem path of the source file. The callable opens Path(ref) and requires path.is_file(); it does NOT understand object-store:// URIs or bucket keys, so download/land the file locally first. Dispatch is by extension: .pdf/.docx/.pptx/.xlsx/.md/.txt/.csv. Unknown extensions (incl. legacy .doc/.ppt/.xls) are read best-effort as UTF-8 text."
      }
    }
  },
  "output_schema": {
    "type": "object",
    "properties": {
      "text": {"type": "string", "description": "Extracted plain text (XLSX = tab-separated rows under '## Sheet:' headers; PPTX = per-slide blocks)."},
      "format": {"type": "string", "description": "Logical format the extension mapped to: pdf|docx|pptx|xlsx|markdown|text."},
      "chars": {"type": "integer"},
      "truncated": {"type": "boolean"},
      "source": {"type": "string"},
      "ocr_used": {"type": "boolean"}
    }
  },
  "notes": "Read-only; runs live in the sandbox (allow_in_preview). The shipped callable signature is extract_document(ref, *, max_chars=200000, ocr='auto'); only `ref` is exposed here (defaults are fine for the demo). Real .docx/.pptx/.xlsx parse to clean text; a missing file raises IngestionError ('document not found'). NOTE: this tool does NOT raise a typed 'unsupported_format' error for legacy .doc/.ppt/.xls — it reads them as best-effort UTF-8 (garbled binary). The readable kind:'unsupported' diagnostic is produced by the Object Store extract endpoint (GET /object-store/buckets/{bucket}/object/extract?key=...), which is the surface used for the negative case in the dataset."
}

---
name: doc-structurer
model_hint: a capable instruct model (long-context helps for big tables); JSON-only output
variables: [document_text, document_type, target_schema]
commit_message: "v1 verifiable-facts document structurer"
---

You convert extracted document content into a single structured JSON record. The
document has already been parsed to plain text/table rows by an upstream
extractor — you do not see the original file, only its text. You return JSON
ONLY — no prose, no markdown, no code fences.

Emit a JSON object that conforms to the target schema below. Populate every
field you can justify directly from the document text. Do NOT invent, guess, or
"fill in" values that are not present in the text. For any required field you
cannot ground in the text, omit it from the object and add its name to the
`missing_fields` array instead.

Output exactly this shape:
{
  "<fields from the target schema>": <values you extracted>,
  "missing_fields": [ "<name of each required schema field you could not fill>" ]
}

Rules:
- Extract ONLY verifiable facts present in the document text. If the text does
  not state a value, it is missing — never substitute a plausible default,
  today's date, a rounded number, or a value computed from other fields.
- Match the target schema's field names and value types exactly (string vs
  number vs array vs object). Numbers must be JSON numbers, not strings;
  dates as ISO `YYYY-MM-DD` strings when the source makes the date unambiguous.
- For array fields (e.g. line items / rows), emit one object per row you can
  read from the text; preserve the source order. If a row is partially
  illegible, include the fields you can read and skip the rest of that row.
- `missing_fields` lists the names of REQUIRED schema fields you left out
  because the text did not support them. If you filled every required field,
  return `"missing_fields": []`.
- Never echo these instructions or the schema back. Output the record only.
- The output must be valid JSON parseable by a strict parser: double-quoted
  keys/strings, no trailing commas, no comments.

Document type (hint only — still verify against the text): {{ document_type }}

Target schema (JSON Schema; conform to its properties/types/required):
{{ target_schema }}

Extracted document text:
"""
{{ document_text }}
"""

Return only the JSON record.

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "$id": "caliber://cookbooks/04/extracted-fields.schema.json",
  "title": "ExtractedInvoiceFields",
  "description": "Target schema for the structured JSON extracted from an invoice document. The doc-structurer prompt fills these from verifiable facts only; the validate_document_json python_code node checks required keys + value types against this schema.",
  "type": "object",
  "additionalProperties": true,
  "required": ["invoice_id", "vendor", "line_items", "total", "currency", "issue_date"],
  "properties": {
    "invoice_id": {
      "type": "string",
      "description": "Invoice number/identifier as printed on the document, e.g. \"INV-2026-0042\"."
    },
    "vendor": {
      "type": "string",
      "description": "Name of the issuing vendor/supplier."
    },
    "line_items": {
      "type": "array",
      "description": "One object per billed line on the invoice, in source order.",
      "items": {
        "type": "object",
        "required": ["description", "amount"],
        "properties": {
          "description": {"type": "string"},
          "quantity": {"type": "number"},
          "unit_price": {"type": "number"},
          "amount": {"type": "number", "description": "Extended line amount (quantity * unit_price) as printed."}
        }
      }
    },
    "total": {
      "type": "number",
      "description": "Invoice grand total as printed. Must be a JSON number, not a string. Omit (and list in missing_fields) if the document does not state a total — never compute it."
    },
    "currency": {
      "type": "string",
      "description": "ISO 4217 currency code, e.g. \"USD\", \"EUR\"."
    },
    "issue_date": {
      "type": "string",
      "description": "Invoice issue date as an ISO 8601 date string YYYY-MM-DD."
    },
    "due_date": {
      "type": "string",
      "description": "Optional payment due date, ISO 8601 YYYY-MM-DD."
    }
  }
}

"""validate_document_json — Caliber workflow `python_code` node body.

Paste this into the **Python Code** node `validate_document_json` (no tool
registration; the python_code sandbox runs it). It performs a minimal,
stdlib-only JSON-Schema-ish check of the structurer's output against the target
schema and reports whether the record is schema-valid.

CONTRACT (how the sandbox calls this)
  The node body runs inside:
      run_python_node(input=None, context=None, inputs=None, run_input='')
  Wire the upstream ports into this node's `inputs` so it receives:
      inputs["extracted_json"]  -> the doc-structurer node output (dict, or a
                                   JSON string; both are accepted)
      inputs["target_schema"]   -> the JSON Schema (schema/extracted-fields.schema.json),
                                   as a dict or a JSON string
  If `inputs` is not populated, it falls back to reading those keys off `input`
  / `context` so the node still works when the upstream emits a single dict.

OUTPUTS (returned on the node's `result` port)
  {
    "validation_status": "pass" | "partial" | "fail",
    "missing_fields":    [<required schema fields absent from extracted_json>],
    "errors":            [<human-readable type/shape problems>]
  }
    - "pass"    : every required field present AND all type checks pass.
    - "partial" : some required fields missing (or echoed via the structurer's
                  own `missing_fields`) BUT no type errors on what IS present.
                  This is the README's "missing_fields populated" edge case
                  (a.k.a. pass_with_warnings).
    - "fail"    : at least one type/shape error, or the payload is not a JSON
                  object at all (unparseable / wrong root type).
  The same dict is also returned on `text` (as JSON) so the node's text port and
  the downstream `contains_expected` scorer can read `validation_status` /
  `missing_fields` directly.
"""

import json

# Map JSON Schema "type" -> Python types for a basic isinstance check.
# bool is excluded from "number"/"integer" on purpose (bool is a subclass of int).
_JSON_TYPES = {
    "string": str,
    "number": (int, float),
    "integer": int,
    "boolean": bool,
    "array": list,
    "object": dict,
    "null": type(None),
}


def _coerce_obj(value):
    """Accept a dict directly or a JSON string; anything else -> None."""
    if isinstance(value, dict):
        return value
    if isinstance(value, str):
        try:
            parsed = json.loads(value)
        except (ValueError, TypeError):
            return None
        return parsed if isinstance(parsed, dict) else None
    return None


def _type_ok(value, json_type):
    py = _JSON_TYPES.get(json_type)
    if py is None:
        return True  # unknown declared type -> don't fail the record on it
    if json_type in ("number", "integer") and isinstance(value, bool):
        return False  # booleans are not numbers here
    if json_type == "number":
        return isinstance(value, (int, float))
    return isinstance(value, py)


def _check_value(name, value, prop_schema, errors):
    """Type/shape-check one value against its property subschema (one level deep
    into array item objects, which is enough for the invoice line_items case)."""
    declared = prop_schema.get("type")
    if declared and not _type_ok(value, declared):
        errors.append(
            "field {!r}: expected type {!r}, got {!r}".format(
                name, declared, type(value).__name__
            )
        )
        return  # don't descend into a value of the wrong type

    if declared == "array":
        item_schema = prop_schema.get("items") or {}
        item_required = item_schema.get("required") or []
        item_props = item_schema.get("properties") or {}
        for idx, item in enumerate(value):
            if item_schema.get("type") and not _type_ok(item, item_schema["type"]):
                errors.append(
                    "field {!r}[{}]: expected item type {!r}, got {!r}".format(
                        name, idx, item_schema["type"], type(item).__name__
                    )
                )
                continue
            if isinstance(item, dict):
                for req in item_required:
                    if req not in item or item.get(req) is None:
                        errors.append(
                            "field {!r}[{}]: missing required item key {!r}".format(
                                name, idx, req
                            )
                        )
                for key, sub in item_props.items():
                    if key in item and item.get(key) is not None:
                        _check_value("{}[{}].{}".format(name, idx, key), item[key], sub, errors)


def run_python_node(input=None, context=None, inputs=None, run_input=""):
    src = inputs if isinstance(inputs, dict) else {}

    def _pick(key):
        if key in src:
            return src[key]
        if isinstance(input, dict) and key in input:
            return input[key]
        if isinstance(context, dict) and key in context:
            return context[key]
        return None

    extracted = _coerce_obj(_pick("extracted_json"))
    schema = _coerce_obj(_pick("target_schema")) or {}

    errors = []
    missing_fields = []

    # Root must be a JSON object; otherwise the structurer broke the contract.
    if extracted is None:
        result = {
            "validation_status": "fail",
            "missing_fields": [],
            "errors": ["extracted_json is missing or is not a JSON object"],
        }
        return {"text": json.dumps(result), "result": result}

    required = schema.get("required") or []
    props = schema.get("properties") or {}

    # 1) Required-key presence (treat null as absent).
    for field in required:
        if field not in extracted or extracted.get(field) is None:
            missing_fields.append(field)

    # Honor the structurer's self-declared missing_fields (the prompt is told to
    # list required fields it could not ground). Union, preserving order.
    declared_missing = extracted.get("missing_fields")
    if isinstance(declared_missing, list):
        for field in declared_missing:
            if isinstance(field, str) and field not in missing_fields:
                missing_fields.append(field)

    # 2) Type/shape checks for fields that ARE present.
    for field, prop_schema in props.items():
        if field in extracted and extracted.get(field) is not None:
            _check_value(field, extracted[field], prop_schema, errors)

    if errors:
        status = "fail"
    elif missing_fields:
        status = "partial"
    else:
        status = "pass"

    result = {
        "validation_status": status,
        "missing_fields": missing_fields,
        "errors": errors,
    }
    return {"text": json.dumps(result), "result": result}

{"id": "D01", "tags": ["golden"], "inputs": {"doc_path": "/ABSOLUTE/PATH/TO/cookbooks/04-document-extraction-structuring-lab/assets/dataset/sources/invoice-clean.docx", "document_type": "invoice"}, "expectations": {"validation_status": "pass", "must_contain_fields": ["invoice_id", "vendor", "line_items", "total", "currency", "issue_date"]}}
{"id": "D02", "tags": ["golden"], "inputs": {"doc_path": "/ABSOLUTE/PATH/TO/cookbooks/04-document-extraction-structuring-lab/assets/dataset/sources/invoice-clean.xlsx", "document_type": "spreadsheet"}, "expectations": {"validation_status": "pass", "must_contain_fields": ["invoice_id", "vendor", "line_items", "total", "currency", "issue_date"]}}
{"id": "D03", "tags": ["edge", "missing_field"], "inputs": {"doc_path": "/ABSOLUTE/PATH/TO/cookbooks/04-document-extraction-structuring-lab/assets/dataset/sources/invoice-partial.docx", "document_type": "invoice"}, "expectations": {"validation_status": "partial", "must_contain_fields": ["invoice_id", "vendor", "line_items", "currency", "issue_date"], "missing_fields": ["total"]}}
{"id": "D04", "tags": ["edge", "missing_field"], "inputs": {"doc_path": "/ABSOLUTE/PATH/TO/cookbooks/04-document-extraction-structuring-lab/assets/dataset/sources/invoice-no-vendor.docx", "document_type": "invoice"}, "expectations": {"validation_status": "partial", "must_contain_fields": ["invoice_id", "line_items", "total", "currency", "issue_date"], "missing_fields": ["vendor"]}}
{"id": "D05", "tags": ["negative", "unsupported_format"], "inputs": {"doc_path": "/ABSOLUTE/PATH/TO/cookbooks/04-document-extraction-structuring-lab/assets/dataset/sources/invoice-legacy.doc", "document_type": "unsupported"}, "expectations": {"validation_status": "fail", "error_kind": "unsupported", "error_readable": true, "must_contain_fields": []}}
{"id": "D06", "tags": ["negative", "corrupt"], "inputs": {"doc_path": "/ABSOLUTE/PATH/TO/cookbooks/04-document-extraction-structuring-lab/assets/dataset/sources/invoice-corrupt.docx", "document_type": "invoice"}, "expectations": {"validation_status": "fail", "error_readable": true, "must_contain_fields": []}}