One JSON object per line, one contract for the whole batch
JSONL stores a separate JSON value on each line. This recipe accepts one event object per line, validates every event and returns outcome counts and duration totals by worker. It processes small inline text with Python's standard library, then prints one summary only after the complete batch passes.
The recipe allows LF or CRLF line endings and an optional final newline. It rejects blank or whitespace-only records, including an extra empty line at the end. Limits are 16 KiB of UTF-8 text, 100 records and 1,024 bytes per record excluding its line ending. A bare carriage return is rejected.
Scroll horizontally to see every column.
| Field | Accepted value |
|---|---|
| event_id | Unique EVT- followed by exactly three ASCII digits |
| worker | 1–24 ASCII characters: lowercase letter first, then lowercase letters, digits or hyphens |
| outcome | Exactly completed or failed |
| duration_ms | Integer from 0 to 120000; no booleans, fractions or numeric strings |
Each event contains exactly those four fields. An array, null, missing field or unknown field rejects the entire input. This is a specific event contract, not a general JSON Schema validator or a live worker telemetry API.
Keep event identity and line numbers separate
{"event_id":"EVT-301","worker":"alpha","outcome":"completed","duration_ms":1250}{"event_id":"EVT-302","worker":"beta","outcome":"failed","duration_ms":2000}{"event_id":"EVT-303","worker":"alpha","outcome":"completed","duration_ms":750}{"event_id":"EVT-304","worker":"beta","outcome":"completed","duration_ms":1000}Alpha has two completed events totaling 2,000 ms. Beta has one failed and one completed event totaling 3,000 ms. The complete dataset contains four events, three completed outcomes, one failure and 5,000 ms. IDs remain unique across all lines; a repeated ID rejects the batch rather than silently deduplicating it.
A JSON string can contain an escaped \n; that escape remains inside one JSONL record. Pretty-printing an event across several physical lines breaks this recipe's one-object-per-line contract. The script iterates io.StringIO, then decodes each complete line with json.loads.
Reject ambiguous JSON before checking event fields
import ioimport jsonimport reimport sysJSONL_TEXT = "{\"event_id\":\"EVT-301\",\"worker\":\"alpha\",\"outcome\":\"completed\",\"duration_ms\":1250}\n{\"event_id\":\"EVT-302\",\"worker\":\"beta\",\"outcome\":\"failed\",\"duration_ms\":2000}\n{\"event_id\":\"EVT-303\",\"worker\":\"alpha\",\"outcome\":\"completed\",\"duration_ms\":750}\n{\"event_id\":\"EVT-304\",\"worker\":\"beta\",\"outcome\":\"completed\",\"duration_ms\":1000}\n"FIELDS = {"event_id", "worker", "outcome", "duration_ms"}def unique_object(pairs): result = {} for key, value in pairs: if key in result: raise ValueError("Duplicate JSON key.") result[key] = value return resultdef reject_number(value): raise ValueError("Only integer JSON numbers are accepted.")def analyze(text): if not isinstance(text, str) or not text or len(text.encode("utf-8")) > 16 * 1024: raise ValueError("Input must be JSONL text within 16 KiB.") if "\r" in text.replace("\r\n", ""): raise ValueError("Use LF or CRLF line endings.") seen = set() groups = {} outcomes = {"completed": 0, "failed": 0} total_duration = 0 count = 0 with io.StringIO(text, newline="") as stream: for line_number, line in enumerate(stream, 1): prefix = "Line " + str(line_number) + ": " if not line.strip(): raise ValueError(prefix + "blank records are not allowed.") if len(line.rstrip("\r\n").encode("utf-8")) > 1024: raise ValueError(prefix + "record exceeds 1024 bytes.") if count >= 100: raise ValueError(prefix + "input exceeds 100 records.") try: event = json.loads(line, object_pairs_hook=unique_object, parse_constant=reject_number, parse_float=reject_number) except (ValueError, RecursionError): raise ValueError(prefix + "invalid JSON, duplicate key or noninteger number.") from None if not isinstance(event, dict) or set(event) != FIELDS: raise ValueError(prefix + "fields do not match the contract.") event_id = event["event_id"] if not isinstance(event_id, str) or not re.fullmatch(r"EVT-[0-9]{3}", event_id) or event_id in seen: raise ValueError(prefix + "invalid or duplicate ID.") worker = event["worker"] if not isinstance(worker, str) or not re.fullmatch(r"[a-z][a-z0-9-]{0,23}", worker): raise ValueError(prefix + "invalid worker label.") outcome = event["outcome"] if not isinstance(outcome, str) or outcome not in outcomes: raise ValueError(prefix + "outcome must be completed or failed.") duration = event["duration_ms"] if type(duration) is not int or not 0 <= duration <= 120_000: raise ValueError(prefix + "duration_ms must be an integer from 0 to 120000.") seen.add(event_id) group = groups.setdefault(worker, {"event_count": 0, "completed": 0, "failed": 0, "duration_ms": 0}) group["event_count"] += 1 group[outcome] += 1 group["duration_ms"] += duration outcomes[outcome] += 1 total_duration += duration count += 1 # Nothing is printed until the complete input has passed validation. return { "event_count": count, "total_duration_ms": total_duration, "outcomes": outcomes, "workers": [dict(worker=name, **groups[name]) for name in sorted(groups)], }if __name__ == "__main__": try: print(json.dumps(analyze(JSONL_TEXT), ensure_ascii=False, sort_keys=True, allow_nan=False)) except (ValueError, UnicodeError) as error: message = "Input must be UTF-8 text." if isinstance(error, UnicodeError) else str(error) print(message, file=sys.stderr) sys.exit(1)python3 --versionpython3 validate-jsonl.pyPython's default JSON decoder accepts repeated object keys and nonstandard NaN or infinity constants. This recipe rejects repeated keys with object_pairs_hook and nonfinite constants with parse_constant. It also rejects fraction and exponent notation with parse_float because its only numeric field must be an integer. JSON decoder behavior.
A failure prints a line number and a fixed rule to stderr, exits with code 1 and prints no partial summary. The original parser exception and raw event values are not echoed. Validation can occur after earlier events have accumulated internally; their totals never reach stdout when a later line fails.
Reconcile outcome counts and per-worker durations
{ "event_count": 4, "total_duration_ms": 5000, "outcomes": { "completed": 3, "failed": 1 }, "workers": [ { "worker": "alpha", "event_count": 2, "completed": 2, "failed": 0, "duration_ms": 2000 }, { "worker": "beta", "event_count": 2, "completed": 1, "failed": 1, "duration_ms": 3000 } ]}The outcome counts add to event_count. Per-worker event counts also sum to four, and their duration_ms values sum to total_duration_ms. A failed event still contributes its duration, which records elapsed work rather than successful work only. These synthetic totals do not establish wall-clock throughput when events overlap.
import assert from "node:assert/strict"import { readFileSync } from "node:fs"const expected = { "event_count": 4, "total_duration_ms": 5000, "outcomes": { "completed": 3, "failed": 1 }, "workers": [ { "worker": "alpha", "event_count": 2, "completed": 2, "failed": 0, "duration_ms": 2000 }, { "worker": "beta", "event_count": 2, "completed": 1, "failed": 1, "duration_ms": 3000 } ]}try { const actual = JSON.parse(readFileSync("result.json", "utf8")) assert.deepStrictEqual(actual, expected) console.log("Fixture result matches.")} catch { console.error("Fixture result did not match the expected contract.") process.exitCode = 1}python3 validate-jsonl.py > result.json && node verify-result.mjsThis exact verifier is for the synthetic fixture. For another event batch, validate the output schema and these count and duration invariants. Keep zero-event input an explicit error rather than reporting a misleading empty success.
Run the validator as a bounded Python job
Follow the quickstart to connect an online Docker worker and create a workspace key. Save the shared local client from JavaScript code execution as run-job.mjs, beside validate-jsonl.py and verify-result.mjs. The text is embedded in the submitted Python source, rather than downloaded or uploaded as a separate file.
# CPUOS_API_KEY is already set in your trusted client environment.# Retain this intent key and source when recovering an uncertain submission.export CPUOS_IDEMPOTENCY_KEY="$(node -p 'crypto.randomUUID()')"node run-job.mjs python validate-jsonl.py > result.json && node verify-result.mjsThe shared client requests 1 CPU, 256 MiB and a 30-second timeout. It parses stdout only after a completed job, exit code 0, no execution error and no truncated output. An accepted submission is not a validated result. Use the Python API guide for terminal states and structured outputs for your application contract.
cpuOS runs trusted team code on your restricted Docker worker, with no job network, package installation, file-transfer API or persistent session. Source and output pass through the EU-hosted control plane; you choose the worker location. Keep credentials outside the source and the complete submitted program within 64 KiB.
Make rejection and quarantine separate contracts
- Test a malformed second line after a valid first line and require empty stdout on failure.
- Test duplicate object keys separately from duplicate event IDs; they are different ambiguities.
- Keep line-ending and blank-record rules explicit when adapting a producer's export.
- If your application intentionally quarantines bad records, return an incomplete status and approved error metadata instead of presenting this all-or-nothing summary as complete.
Use this validator for a reviewed n8n step or LangChain tool. For one JSON document rather than line-delimited events, see validate and transform JSON. For tabular exports with quoted newlines, use the CSV parser, whose record boundaries follow different rules.