Ingestion (ingestion.*)

One of the poq.toml spec sections.

Everything before review: declare sources, optional joins, and the field projection that becomes each datapoint row. Turn uploaded files into datapoint rows: declare sources, optional joins, and which columns each task carries.

• ingestion.sources (required) — Where your data lives and how Sapien reads it.
• ingestion.joins (optional) — Combine separate sources before field projection.
• ingestion.fields (required) — Name each column on a task and say which file it came from (e.g. title = "findings.title").

ingestion.sources

[[ingestion.sources]] tells Sapien which files contain your project's data and how to read them. Every batch of files is its own [[ingestion.sources]] block.

One Primary Source: Sapien builds your task list from the first [[ingestion.sources]] block listed in your TOML. If you want tasks from multiple files or folders, you should group them into this first block using a glob.

• Multiple Markdown files: Use path_glob = "reports/*.md". Every section in every matching file will become a task.
• Multiple JSON files (one item per file): Use path_glob = "{folder_a,folder_b}/*.json". Every JSON file in both folders becomes one task.
• JSON report files with an array of items: Use path_glob = "reports/*.json" with unnest.array_key = "findings" (or whatever key holds the list). Each array element becomes one task — no manual pre-split into per-item files.

If you create separate [[ingestion.sources]] blocks (with different ids), Sapien assumes they are different types of data that need to be joined together (like joining a CSV of labels to a folder of images). Data in secondary inputs will only appear in your tasks if you explicitly join them to the first input.

[[ingestion.sources]]
id = "findings"
type = "json"
path_glob = "findings/*.json"

The id is the label you pick for this batch. Wire columns in [ingestion.fields] using that id as the prefix:

[ingestion.fields]
title = "findings.title"    # "findings" is the source id; ".title" is the column

With multiple inputs (CSV + image folder), ids disambiguate sources:

[[ingestion.sources]]
id = "labels"
type = "csv"
path = "labels.csv"

[[ingestion.sources]]
id = "images"
type = "file_collection"
path_glob = "images/*.jpg"

[ingestion.fields]
diagnosis = "labels.diagnosis"

Input fields

Per-type field set

Each type allows a different subset of keys. The compiler rejects forbidden keys with a path-prefixed error (for example inputs[1].path_glob: only valid on file_collection or json inputs).

json and markdown_split accept either path (single file) or path_glob (many files). Setting both is a compile-time error.

JSON array unnest

By default, each uploaded JSON file becomes one review item. Use unnest.array_key on a json input when each file is a report object that wraps an array of review items at a known key — ingest explodes that array into one review item per element (deterministic key-level splitting; no regex, no interim folder).

[[ingestion.sources]]
id               = "findings"
type             = "json"
path_glob        = "reports/*.json"
unnest.array_key = "findings"

Given reports/batch-001.json:

{
  "exportedAt": "2026-06-23T05:15:03Z",
  "findings": [
    { "fingerprint": "F-01", "title": "Unchecked return value", "severity": "medium", "affectedCode": "..." },
    { "fingerprint": "F-02", "title": "Missing access control", "severity": "high", "affectedCode": "..." }
  ]
}

Ingest produces two review items. Each row's columns are the fields of that array element. Root-level scalars outside the array (here exportedAt) are not copied onto the rows — wire only fields that exist on each element. When the input uses path_glob, every row also carries DuckDB's synthetic filename column (the source file's URL), the same as a flat json glob ingest.

Wire [ingestion.fields] using the exact JSON property names on each array element (camelCase preserved):

[ingestion.fields]
id            = "findings.fingerprint"
title         = "findings.title"
severity      = "findings.severity"
affected_code = "findings.affectedCode"

*Required when you declare unnest for this input; omit unnest entirely when each JSON file is already one review item.

Rules:

• Only valid on type = "json". Setting unnest.array_key on csv, file_collection, or markdown_split fails at parse time.
• The value at unnest.array_key must be a JSON array. An empty array yields zero review items from that file.
• Each array element should be a JSON object so its keys can be wired in [ingestion.fields].
• Only each element's own fields become row columns; root-level scalars outside the array are not carried.

The markdown_split input type

Use markdown_split when one or more Markdown files should become many review items. Each regex match in splitter.regex becomes one row on that input.

Write the pattern from header lines in your file — copy the fixed prefix literally, use (?P<name>...) for the parts that change on each row:

Optional splitter.end_regex — stop the section body before the next header when subsections use the same ## level:

Optional [[ingestion.sources.splitter.metadata]] — extract extra columns from a line in the file or in each section.

Repository row (document header table) — string → regex:

| Repository | my-repo |
'^\|\s*Repository\s*\|\s*(?P<repository>[^|]+?)\s*\|'

File line (per section) — string → regex:

- **File**: `src/auth.ts:L42`
'^-\s*\*\*File\*\*:\s*`?(?P<source_file>.+?):L(?P<line_number>\d+)'

[[ingestion.sources]]
id = "audit_findings"
type = "markdown_split"
path_glob = "reports/*.md"
splitter.regex = '^##\s+(?P<id>F-\d+):?\s+(?P<title>.+)
#x27;
splitter.end_regex = '^##\s+'

[[ingestion.sources.splitter.metadata]]
scope = "document"
column = "repository"
regex = '^\|\s*Repository\s*\|\s*(?P<repository>[^|]+?)\s*\|'

Splitter settings

These keys apply only when type = "markdown_split". Regex values use RE2; Sapien adds multiline matching ((?m)) at compile time — do not prefix patterns yourself.

After ingest, each split row exposes columns you can wire in [ingestion.fields] — e.g. named captures from splitter.regex, metadata column names, and built-ins such as body and row_id:

[ingestion.fields]
finding_body = "audit_findings.body"
repo         = "audit_findings.repository"

ingestion.joins

The purpose of [[ingestion.joins]] is to line up related data from different files so they can be treated as a single task.

Use this section only if you declared more than one [[ingestion.sources]] block. It merges your separate data sources into one wider "spreadsheet" before you pick your final columns in [ingestion.fields].

[[ingestion.joins]]
left = "labels"
right = "images"
left_on = "case_id"
right_on = "file_id"
type = "left"

FROM-root order (required for connected joins)

In plain terms: think of building one big spreadsheet by gluing smaller ones together. Start with your main file (one row per task — e.g. labels.csv) and list it first. Each join then glues a new file onto what you already have: left is the file you already have, right is the file you're adding. Always glue new files onto the main one — never the other way around. If you flip it, the tool can't attach your main file and ingest fails.

Ingest builds SQL with the first [[ingestion.sources]] entry as the FROM root (one row per review item — map id from this source). Each [[ingestion.joins]] row adds right as a new table; left must already be in the FROM chain.

Getting this backwards (e.g. left = "images", right = "labels" when labels is primary) produces ingest SQL where the primary table never enters FROM.

# Primary CSV first, then join enrichment
[[ingestion.sources]]
id = "labels"
type = "csv"
path = "labels.csv"

[[ingestion.sources]]
id = "images"
type = "file_collection"
path_glob = "images/*.jpg"

[[ingestion.joins]]
left = "labels"      # primary / already in FROM
right = "images"     # new table
left_on = "image_id"
right_on = "file_id"
type = "left"

ingestion.fields

[ingestion.fields] is one flat table that defines every column on each review item. Each key is the column name used everywhere after ingest; each value is where the data comes from: <source_id>.<column>.

If you used [[ingestion.joins]] to merge multiple sources, values can reference columns from any joined source.

After ingest, those columns are stored on each review item in the database. See The Data Lifecycle.

Every project must include exactly one key named id. That key is the unique identifier for each review task.

Ten fields need about ten lines — not thirty repeating two-key blocks.

Worked example: JSON findings (finding-004.json)

Example file: datasets/test-audit-contract/findings/finding-004.json in poq-monorepo. One JSON file = one review item.

Step 1 — declare the input (gives you the findings. prefix):

[[ingestion.sources]]
id = "findings"
type = "json"
path_glob = "findings/*.json"

Step 2 — map JSON keys to review-item columns. Each row below becomes one line in [ingestion.fields]. The value uses the JSON property name after ingest; the key is what you use everywhere else — and what lands in datapoint.canonical_fields.

[ingestion.fields]
id                = "findings.id"
finding_title     = "findings.title"
description       = "findings.description"
source_path       = "findings.sourcePath"
line_number       = "findings.lineNumber"
proposed_severity = "findings.proposedSeverity"
repository        = "findings.repository"
commit_sha        = "findings.commitSha"

Step 3 — use these keys in later sections. Reference [ingestion.fields] keys only — never raw <source_id>.<column> paths outside that table.

For example, to show the finding's description to a validator:

[[validation.evidence]]
type = "markdown"
ingestion_field = "description"

Or to route tasks based on severity:

[[validators.routes]]
match = { proposed_severity = "low" }
total = 5

You should never reference the original source (like findings.sourcePath or findings.proposedSeverity) outside of [ingestion.fields].

CSV + images example (two inputs, no JSON):

[ingestion.fields]
id         = "findings.case_id"
image_path = "images.path"