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, markdown_split, and video_frames 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 a single Markdown file (or a folder of them) is really many review items stacked in one document — an audit with one section per finding, a catalog with one entry per heading, a postmortem with repeated sections. It is the Markdown path into the same spreadsheet model as CSV rows and JSON elements (see The Data Lifecycle): where a CSV gives one row per line and a JSON array gives one row per element, markdown_split gives one row per section.

How the split works. Sapien scans each file top to bottom. Every line that matches splitter.regex opens a new section; that section runs from its header down to the start of the next matching header — or, if set, to the first splitter.end_regex match — or to end of file. Each section becomes one row / one datapoint, exactly like a CSV line.

What columns a section row has. Every row carries the section's own text as body, plus built-in columns like row_id and report_path, plus one column for each named capture in splitter.regex, plus any columns you pull with metadata. You wire those columns into review-item fields in ingestion.fields.

Write the splitter 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 (Go's regexp): linear-time, but no backreferences or lookaround — anchors (^, $), character classes, and named captures (?P<name>...) are all supported. Sapien adds multiline matching ((?m)) at compile time, so ^ and $ match at line boundaries — do not prefix (?m) yourself. Extracted metadata values are trimmed of surrounding whitespace and backticks, so a table cell like `a1b2c3d4` stores as a1b2c3d4.

Columns available after a split

Each split row exposes three kinds of column, all wired the same way in [ingestion.fields] (<field> = "<input_id>.<column>"):

• Named captures from splitter.regex — e.g. id, title above.
• Metadata columns — every column you declared under [[ingestion.sources.splitter.metadata]].
• Built-ins — always present on every row, listed below.

Referencing a column that does not exist (no such capture, metadata, or built-in) is a compile-time error naming the missing column, so typos fail fast rather than yielding blank fields.

[ingestion.fields]
id           = "audit_findings.row_id"    # built-in — one stable id per section
finding_id   = "audit_findings.id"        # named capture from splitter.regex
finding_body = "audit_findings.body"      # built-in — the section text
repo         = "audit_findings.repository" # metadata column

The video_frames input type

Use video_frames when an uploaded video should become one review item per frame. Upload the raw video with your project folder; Sapien splits it server-side, stores every extracted frame durably, and emits one row per frame. You do no ETL: no frame splitting, no CSV, no re-encoding. Extracted frame sets are content-addressed by the video's SHA-256, so re-ingesting the same bytes (or retrying a failed run) reuses the already-extracted frames.

[[ingestion.sources]]
id   = "frames"
type = "video_frames"
path = "recording.mp4"       # or path_glob = "videos/*.mp4"
sample_fps = 30              # required: 30 fps footage scored at 30 = every frame

video_frames settings

sample_fps is required. The rest are optional, and default to a JPEG frame with a one-second context clip.

There is deliberately no default rate. How densely a video is scored decides both what the audit can catch and what it costs, and neither answer is safe to assume on your behalf — scoring every frame is a choice too, so it gets written down. The frames Sapien keeps are exact native frames carrying their true native indices in frame_index, never a re-timed rate. That is what lets a sampled audit line up with annotation files that are themselves indexed by frame number. Because the step is a whole number of frames, a rate the video's frame rate cannot divide evenly lands on the nearest achievable one — asking for 7 frames per second of 30 fps footage keeps every 4th frame, which is 7.5 per second — and an exact half rounds toward more frames, so a run is never sparser than the rate you asked for.

Columns every frame row exposes

Worked example: per-frame video annotation audit

[[ingestion.sources]]
id   = "frames"
type = "video_frames"
path = "grid_2x2.mp4"
sample_fps = 30

[ingestion.fields]
id        = "frames.frame_id"
frame_url = "frames.frame_url"

[[validation.evidence]]
type  = "image"
title = "This frame"
url   = "frame_url"

That is the whole spec: the reviewer sees the frame and scores it. An 85-second 1080p30 video ingests as exactly 2,550 datapoints, one per frame. The project wizard's preview extracts only the first 25 frames so it answers in seconds; the full extraction happens on the real ingest run after project creation.

Optional evidence for a frame

Two more blocks exist for when the work needs them — not by default, because each one is another thing the reviewer has to look at on every one of those 2,550 items.

The previous frame beside this one catches an annotation that blinks out for a single frame; add it when the rubric is about consistency across frames rather than about one frame on its own.

[ingestion.fields]
prev_frame_url = "frames.prev_frame_url"

[[validation.evidence]]
type  = "image_comparison"
title = "Previous frame vs this frame"
left_image  = "prev_frame_url"
right_image = "frame_url"

A short clip around the moment adds motion; add it when a still is genuinely ambiguous. Bind poster to frame_url so the frame shows at once instead of waiting on the video, and note that the clip only plays if the source is in a format browsers decode (H.264 in an MP4) — extraction reads formats browsers do not, so a clip can extract perfectly and still not play.

[ingestion.fields]
video_url          = "frames.video_url"
clip_start_seconds = "frames.clip_start_seconds"
clip_end_seconds   = "frames.clip_end_seconds"

[[validation.evidence]]
type          = "video_clip"
title         = "One second around this moment"
video_url     = "video_url"
poster        = "frame_url"
start_seconds = "clip_start_seconds"
end_seconds   = "clip_end_seconds"

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"