fix(ci): treat an unreadable PharmCAT report as not assessed

.github/workflows/guard.yml

@@ -279,6 +279,16 @@ jobs:
           grep -E 'planted_failure.*FAILED' "$log" || { cat "$log"; echo "ERROR: the run failed, but not on the planted test"; exit 1; }
           echo "Control: the planted test failed, as it must."
 
+      - name: Control (an unreadable PharmCAT report reported as success must fail)
+        run: |
+          log="${RUNNER_TEMP}/ci-reader-control.log"
+          rc=0
+          cargo test --locked --no-fail-fast -p og-dev --features planted-ci-reader --test ci_readers \
+            unreadable_pharmcat_is_unassessed > "$log" 2>&1 || rc=$?
+          cat "$log"
+          [ "$rc" -ne 0 ] || { echo "ERROR: the planted unreadable-report success passed"; exit 1; }
+          grep -E 'unreadable_pharmcat_is_unassessed.*FAILED' "$log"
+          echo "Control: the unreadable-report success was reported as a failed test."
       - name: Traits controls (diplotype order, REF check and missing phase rows)
         env:
           CARGO_BUILD_JOBS: 3

.github/workflows/nextflow.yml

@@ -119,8 +119,8 @@ jobs:
       - name: Validate nextflow_schema.json and the samplesheet schema
         if: steps.scope.outputs.run == 'true'
         run: |
-          python3 -m json.tool nextflow_schema.json > /dev/null
-          python3 -m json.tool assets/schema_input.json > /dev/null
+          jq -s -e 'length == 1' nextflow_schema.json > /dev/null
+          jq -s -e 'length == 1' assets/schema_input.json > /dev/null
 
       # The stub run below pulls every image of the default tools: the pinned
       # leg ran out of disk in it twice on 2026-11-05 (#300, job 127077180393;
@@ -430,83 +430,10 @@ jobs:
           nextflow run main.nf -profile test_all -stub -ansi-log false -c "${RUNNER_TEMP}/resources.config" \
             --tools vep,manta,cnvpytor,prs --max_cpus 16 --max_memory 64.GB --max_time 48.h \
             --outdir "${RUNNER_TEMP}/resources" -work-dir "${RUNNER_TEMP}/resources-work"
-          python3 - "${RUNNER_TEMP}"/resources/pipeline_info/trace_*.txt "$exec_cpus" "$exec_mem_gb" <<'PY'
-          import csv, sys
-          GB, H = 1024 ** 3, 3600 * 1000
-          exec_cpus, exec_mem = int(sys.argv[2]), int(sys.argv[3]) * GB
-          # process: (cpus, memory, time) of the first attempt
-          want = {
-              'DEEPVARIANT':   (8, 32 * GB, 48 * H),
-              'FASTP':         (8, 8 * GB, 8 * H),
-              'ALIGN_MINIMAP2': (8, 32 * GB, 24 * H),
-              'MANTA':         (8, 16 * GB, 24 * H),
-              'CNVPYTOR':      (4, 8 * GB, 24 * H),
-              'VEP':           (8, 16 * GB, 24 * H),
-              'PRS_PREPARE':   (1, 2 * GB, 2 * H),
-              'PRS_SCORE_SITES': (2, 4 * GB, 4 * H),
-              'PRS':           (4, 24 * GB, 8 * H),
-          }
-          seen, bad = {p: {} for p in want}, 0
-          attempts = all_cpus = all_mem = 0
-          rows = {}
-          with open(sys.argv[1]) as fh:
-              for row in csv.DictReader(fh, delimiter='\t'):
-                  first_row = rows.setdefault(row['task_id'], row)
-                  if first_row is not row:
-                      # The late copy can be built before the exit status was
-                      # set, so it may read '-' there; nothing else may differ.
-                      exits = {first_row['exit'], row['exit']} - {'-'}
-                      if len(exits) > 1 or any(first_row[k] != row[k] for k in row if k != 'exit'):
-                          print(f"FAIL: task {row['task_id']} has two different rows in the trace: {first_row} and {row}")
-                          bad += 1
-                      else:
-                          print(f"{row['name']}: attempt {row['attempt']} is written twice in the trace; counted once")
-                      continue
-                  attempts += 1
-                  all_cpus += int(row['cpus'])
-                  all_mem += int(row['memory'])
-                  proc = row['name'].split(' (')[0].split(':')[-1]
-                  if proc not in want:
-                      continue
-                  got = (int(row['cpus']), int(row['memory']), int(row['time']))
-                  print(f"{row['name']}: attempt {row['attempt']}, exit {row['exit']}, "
-                        f"cpus {got[0]}, memory {got[1] / GB:g} GB, time {got[2] / H:g} h")
-                  seen[proc].setdefault(int(row['attempt']), {}).setdefault(row['name'], []).append(got)
-          for proc, (cpus, mem, time) in want.items():
-              first, second = seen[proc].get(1, {}), seen[proc].get(2, {})
-              unexpected = sorted(set(seen[proc]) - {1, 2})
-              if unexpected:
-                  print(f"FAIL {proc}: attempt(s) {unexpected} in the trace, want only a first attempt and one retry")
-                  bad += 1
-              if not first:
-                  print(f"FAIL {proc}: no first attempt in the trace")
-                  bad += 1
-                  continue
-              for name in sorted(set(first) | set(second)):
-                  n1, n2 = len(first.get(name, [])), len(second.get(name, []))
-                  if n1 != 1 or n2 != 1:
-                      print(f"FAIL {name}: {n1} first and {n2} second attempts in the trace, want one of each")
-                      bad += 1
-              for got in (g for gs in first.values() for g in gs):
-                  if got != (cpus, mem, time):
-                      print(f"FAIL {proc}: first attempt asks for cpus {got[0]}, memory {got[1] / GB:g} GB, "
-                            f"time {got[2] / H:g} h; want {cpus}, {mem / GB:g} GB, {time / H:g} h")
-                      bad += 1
-              for got in (g for gs in second.values() for g in gs):
-                  if got[1] != 2 * mem:
-                      print(f"FAIL {proc}: the retry asks for {got[1] / GB:g} GB, want {2 * mem / GB:g} GB")
-                      bad += 1
-          # Every attempt takes its request once and gives it back once. If
-          # the executor has at least the sum, losing every release still
-          # leaves room for the next task.
-          print(f"{attempts} task attempts ask for {all_cpus} CPUs and {all_mem / GB:g} GB added together; "
-                f"the executor is told it has {exec_cpus} CPUs and {exec_mem / GB:g} GB")
-          if all_cpus > exec_cpus or all_mem > exec_mem:
-              print("FAIL: executor.cpus or executor.memory of this step is below that sum; raise exec_cpus / exec_mem_gb")
-              bad += 1
-          print(f"{bad} failure(s)")
-          sys.exit(1 if bad else 0)
-          PY
+          traces=("${RUNNER_TEMP}"/resources/pipeline_info/trace_*.txt)
+          [ "${#traces[@]}" -eq 1 ] && [ -f "${traces[0]}" ] || { echo "FAIL: expected exactly one resource trace"; exit 2; }
+          rust/target/release/og dev nf-trace requests "${traces[0]}" --want scripts/ci/nf-trace-requests.tsv \
+            --executor-cpus "$exec_cpus" --executor-mem-gb "$exec_mem_gb"
 
       # The step above fakes the kill with beforeScript, so it cannot see how
       # ALIGN_MINIMAP2's own pipe reports one. This runs the module's real
@@ -574,28 +501,8 @@ jobs:
             --max_cpus 16 --max_memory 64.GB --outdir "$t/out" -work-dir "$t/work" || rc=$?
           echo "nextflow exited ${rc}"
           cat "$t/trace.txt"
-          python3 - "$t/trace.txt" "$rc" <<'PY'
-          import csv, sys
-          GB = 1024 ** 3
-          # One row per task_id: the trace can list a finished task twice
-          # (the retry step above says why), and the first row is the one
-          # written when the task completed.
-          # A repeated row must match the first; only its exit may read '-'.
-          rows = {}
-          for r in csv.DictReader(open(sys.argv[1]), delimiter='\t'):
-              first = rows.setdefault(r['task_id'], r)
-              exits = {first['exit'], r['exit']} - {'-'}
-              if len(exits) > 1 or any(first[k] != r[k] for k in r if k != 'exit'):
-                  print(f"FAIL: task {r['task_id']} has two different rows in the trace: {first} and {r}")
-                  sys.exit(1)
-          rows = [r for r in rows.values() if 'ALIGN_MINIMAP2' in r['name']]
-          got = [(int(r['attempt']), r['exit'], int(r['memory']) // GB) for r in rows]
-          want = [(1, '137', 32), (2, '0', 64)]
-          print(f"ALIGN_MINIMAP2 attempts (attempt, exit, GB): {got}; want {want}")
-          ok = sorted(got) == want and sys.argv[2] == '0'
-          print('OK' if ok else 'FAIL: a kill inside the pipe did not reach Nextflow as exit 137 and a retry')
-          sys.exit(0 if ok else 1)
-          PY
+          rust/target/release/og dev nf-trace attempts "$t/trace.txt" --name-glob '*ALIGN_MINIMAP2*' \
+            --want '1:137:32,2:0:64' --nextflow-exit "$rc"
 
       # PRS runs pgsc_calc, a Nextflow pipeline of its own, and gives it the
       # task's memory as --max_memory, the most one pgsc_calc task may have. A
@@ -696,26 +603,10 @@ jobs:
             --max_cpus 16 --max_memory 64.GB --outdir "$t/out" -work-dir "$t/work" || rc=$?
           echo "nextflow exited ${rc}"
           cat "$t/trace.txt"
-          python3 - "$t/trace.txt" "$rc" <<'PY'
-          import csv, sys
-          GB = 1024 ** 3
-          # One row per task_id, as in the ALIGN_MINIMAP2 step above.
-          rows = {}
-          for r in csv.DictReader(open(sys.argv[1]), delimiter='\t'):
-              first = rows.setdefault(r['task_id'], r)
-              exits = {first['exit'], r['exit']} - {'-'}
-              if len(exits) > 1 or any(first[k] != r[k] for k in r if k != 'exit'):
-                  print(f"FAIL: task {r['task_id']} has two different rows in the trace: {first} and {r}")
-                  sys.exit(1)
-          rows = list(rows.values())
-          want, ok = [(1, '137', 24), (2, '0', 48)], sys.argv[2] == '0'
-          for sample in ('sigkill', 'pool'):
-              got = sorted((int(r['attempt']), r['exit'], int(r['memory']) // GB) for r in rows if r['name'] == f'PRS ({sample})')
-              print(f"PRS ({sample}) attempts (attempt, exit, GB): {got}; want {want}")
-              ok = ok and got == want
-          print('OK' if ok else 'FAIL: a memory kill inside pgsc_calc did not reach Nextflow as exit 137 and a retry that passes')
-          sys.exit(0 if ok else 1)
-          PY
+          for sample in sigkill pool; do
+            rust/target/release/og dev nf-trace attempts "$t/trace.txt" --name-glob "PRS ($sample)" \
+              --want '1:137:24,2:0:48' --nextflow-exit "$rc"
+          done
 
       # A Nextflow run that fails, or that its step's timeout cuts, leaves its
       # log in the checkout as .nextflow.log (older runs of the job as

CHANGELOG.md

@@ -4,6 +4,8 @@ Each release's notes. From 1.0.0 on, a major or minor release has hand-written n
 
 ## Unreleased
 
+- The shared `og-io` line reader bounds memory by the longest record for CR-only files as well as LF and CRLF files. Smoke coordinates require signed integers, so decimal and nonfinite coordinates fail. Trace integer cells accept whitespace, signs and underscores but deliberately refuse Unicode digits and negative resources. Integers outside the target range fail. Duplicate task rows retain concrete exits. Smoke JSON checks require one document, numeric counts and copy numbers, array catalogs and string genotypes. Numeric depth strings stay supported; booleans are refused. The array and genotype rules deliberately reject malformed values Python accepted. Smoke fixtures build a missing default `og` on first use. Setup output goes to stderr so captured PharmCAT counts and CSV values stay clean.
+- CI readers now use `og dev pharmcat diplotypes`, `nf-trace` and `smoke`. Unreadable PharmCAT reports fail with exit 2 and are marked not assessed. Resource checks refuse trace globs and malformed numeric cells. The smoke mini-reference uses a fixed Rust seed for the planted bases. Gzip smoke reads have deterministic headers. Smoke table checks report concise property results.
 - Step 42 reads traits and genotype rules with `og traits table`, exports WhatsHap sites with `og traits phase-sites`, and checks the catalogs with `og traits check`. The replaced Python evaluator is removed. Report columns and readings stay the same. Genotypes outside the catalog are streamed past, and write errors now report a controlled error with the path instead of a traceback. Both tables are staged before either is published. A staging failure or a directory destination leaves both previous tables intact. Each file is then replaced by a separate atomic rename. The pair is not a transaction: a failure of the second rename can leave the first table updated. Catalogs and output remain plain TSV; quotes never hide a tab or bypass validation.
 - The unused Python copy-number engine is removed. Its historical catalog tests materialise a pinned test oracle and keep testing the Rust commands.
 

docs/rust.md

@@ -111,6 +111,18 @@ To add a primitive: write it in `pycompat` with unit tests, add an operation to
 
 [Testing](testing.md) has the rest of CI.
 
+## CI readers
+
+`og dev pharmcat diplotypes REPORT` prints sorted, unique `GENE\tA1/A2` rows from flat or source-nested reports. `--called` prints the called count and names on stderr. `--called-lines` prints one called gene per line for e2e checks. An unreadable report exits 2 without a count. A valid empty report succeeds with a zero count.
+
+`og dev nf-trace requests TRACE --want FILE --executor-cpus N --executor-mem-gb N` checks initial requests, doubled retry memory and the executor budget. The expectation table is [nf-trace-requests.tsv](https://github.com/GeiserX/OwnGenome/blob/main/scripts/ci/nf-trace-requests.tsv). `og dev nf-trace attempts TRACE --name-glob PATTERN --want '1:137:32,2:0:64' --nextflow-exit RC` checks a retry sequence. Both accept one explicit path, refuse globs, parse quoted TSV records and reject conflicting duplicates. Identical duplicate task IDs count once. Exit 1 means the check found a mismatch. Exit 2 means input or arguments could not be assessed.
+
+`og dev smoke` supplies `mini-ref SRC ORIG MINI POS LEN`, `long-reads HAP1 HAP2 OUT.gz`, `eh-catalog REF OUT END`, `csv-cell FILE KEY=VALUE[,KEY=VALUE] COLUMN` and `check KIND FILE`. Check kinds are `hap-ibd`, `ibdmix`, `rfmix`, `eigen` and `parascopy`. Mini-reference bases use a fixed Rust seed. Long-read gzip headers are deterministic. These synthetic fixtures do not change pipeline output files. Smoke references are limited to 1,003,000 bases. CSV lookups and table checks read one record at a time.
+
+CI integer cells accept surrounding whitespace, signs and digit separators. Trace cells deliberately refuse Unicode digits and negative resource values. Smoke coordinates accept Unicode decimal digits and use signed 128-bit integers. Integers beyond the target range fail. Decimal and nonfinite coordinates fail. Duplicate trace rows retain the first concrete exit status and reject later conflicts.
+
+The shared `og-io` reader consumes LF, CRLF and lone CR records with storage bounded by the longest line. Smoke JSON checks require one document and typed fields. Catalogs must be arrays and genotypes must be nonempty strings. This deliberately rejects malformed values that Python truthiness or `len()` accepted. Numeric depth strings remain supported; booleans are refused. Smoke fixtures and assertions build `og` on first use if the default binary is missing. An explicit `OG` is used as supplied.
+
 Step 42 uses `og traits table`, `og traits phase-sites` and `og traits check`.
 The readers retain only catalog sites. The trait and genotype-rule output
 columns and readings stay the same. `bin/traits_report.py` keeps report

rust/README.md

@@ -35,6 +35,7 @@ cargo build --release && ../tests/rust/pycompat-diff.sh target/release/og   # py
 
 A binary from a plain `cargo build` is stamped `unknown`, and `rust/og.sh check` refuses it: only `scripts/setup.sh --og` stamps the tree hash the pipeline checks.
 
+CI readers also run through `og dev pharmcat diplotypes`, `og dev nf-trace requests|attempts` and `og dev smoke`. See [the Rust command notes](../docs/rust.md#ci-readers).
 `og traits table`, `og traits phase-sites` and `og traits check` read step 42's
 curated catalogs and synthetic or bcftools genotype input. Their tests live
 in `og-steps::traits` and the `traits` parity cases.

rust/crates/og-dev/Cargo.toml

@@ -6,6 +6,10 @@ edition.workspace = true
 license.workspace = true
 publish.workspace = true
 
+# Only the CI red control enables this fault. Normal builds keep it off.
+[features]
+planted-ci-reader = []
+
 [dependencies]
 pycompat.workspace = true
 og-io.workspace = true

rust/crates/og-dev/src/lib.rs

@@ -21,16 +21,19 @@
 pub mod build;
 pub mod contract;
 pub mod helpers;
+pub mod nf_trace;
 pub mod parity;
+pub mod pharmcat;
 pub mod remnants;
 pub mod report_only;
 pub mod schema;
 pub mod shlex;
+pub mod smoke;
 mod zip;
 
 use std::io::Write;
 
-const USAGE: &str = "usage: og dev parity-compare KIND ORACLE NEW\n       og dev parity-compare --self-test\n       og dev contract --static [--repo DIR] [--contract FILE]\n       og dev contract --outputs DIR --leg bash|nextflow --sample S [--observe]\n       og dev schema SCHEMA.json DOC.json\n       og dev report-only [--repo DIR]\n       og dev remnants [--list FILE] [--root DIR]\n       og dev helpers --list|--static|--images|--orad-calls [--root DIR]\n       og dev helpers --orad --help-file FILE [--orad-version V] [--root DIR]\n       og dev build cn-intervals|callability|sv-rule-regions [args]\n";
+const USAGE: &str = "usage: og dev parity-compare KIND ORACLE NEW\n       og dev parity-compare --self-test\n       og dev contract --static [--repo DIR] [--contract FILE]\n       og dev contract --outputs DIR --leg bash|nextflow --sample S [--observe]\n       og dev schema SCHEMA.json DOC.json\n       og dev report-only [--repo DIR]\n       og dev remnants [--list FILE] [--root DIR]\n       og dev pharmcat diplotypes REPORT [--called|--called-lines]\n       og dev nf-trace requests|attempts TRACE [options]\n       og dev smoke mini-ref|long-reads|eh-catalog|csv-cell|check [args]\n       og dev helpers --list|--static|--images|--orad-calls [--root DIR]\n       og dev helpers --orad --help-file FILE [--orad-version V] [--root DIR]\n\n       og dev build cn-intervals|callability|sv-rule-regions [args]";
 
 /// Python's `str(OSError)`: `[Errno 2] No such file or directory: 'path'`.
 pub(crate) fn py_os_error(e: &std::io::Error, path: &str) -> String {
@@ -65,6 +68,9 @@ pub fn main(args: &[String]) -> u8 {
         Some("contract") => run(contract::cli, &args[1..]),
         Some("schema") => run(schema::cli, &args[1..]),
         Some("report-only") => run(report_only::cli, &args[1..]),
+        Some("pharmcat") => run(pharmcat::cli, &args[1..]),
+        Some("nf-trace") => run(nf_trace::cli, &args[1..]),
+        Some("smoke") => run(smoke::cli, &args[1..]),
         Some("build") => run(build::cli, &args[1..]),
         Some("helpers") => run(helpers::cli, &args[1..]),
         Some("-h" | "--help") => {