From 7b3206d38557f866bfe563e29b98be7c8b7609ab Mon Sep 17 00:00:00 2001 From: Paul Newling Date: Wed, 10 Jun 2026 09:03:09 -0700 Subject: [PATCH 01/10] docs: add parallel-compute implementation plan for sequence-properties --- ...10-sequence-properties-parallel-compute.md | 1028 +++++++++++++++++ ...-properties-parallel-compute.md.tasks.json | 46 + 2 files changed, 1074 insertions(+) create mode 100644 docs/superpowers/plans/2026-06-10-sequence-properties-parallel-compute.md create mode 100644 docs/superpowers/plans/2026-06-10-sequence-properties-parallel-compute.md.tasks.json diff --git a/docs/superpowers/plans/2026-06-10-sequence-properties-parallel-compute.md b/docs/superpowers/plans/2026-06-10-sequence-properties-parallel-compute.md new file mode 100644 index 0000000..21302b4 --- /dev/null +++ b/docs/superpowers/plans/2026-06-10-sequence-properties-parallel-compute.md @@ -0,0 +1,1028 @@ +# Sequence-Properties Parallel Compute Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers-extended-cc:subagent-driven-development (recommended) or superpowers-extended-cc:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Let the `sequence-properties` block compute a huge dataset on the first run by parallelizing the Python property kernel across CPU cores, behind a frozen characterization net that proves the output is byte-identical to today's. + +**Architecture:** Three phases. **Phase A (characterization, do first):** freeze a full-output golden master + structural-invariant scale test from the *current* sequential code, so any behavior change in later phases fails loudly. **Phase B (parallelism, option #1):** map the existing pure per-row compute functions across a `ProcessPoolExecutor`, assemble results in input order, keep the existing sort + CID-quantization boundary — so worker count never changes a byte; bump the workflow's `.cpu()`/`.mem()`. **Phase C (streaming, option #7, OPTIONAL):** stream sorted input in batches to bound peak memory; gated on a measurement decision after Phase B. + +**Tech Stack:** Python 3.12, `polars` (TSV I/O + frame assembly), `biopython` (property math), `concurrent.futures.ProcessPoolExecutor` (parallelism), `pytest` + `uv` + `ruff` (test/lint, run from `software/`), Tengo workflow (`exec.builder()` resource requests). + +--- + +## Before You Start (read once, do not skip) + +1. **Worktree + branch.** Do not work on `main` (workspace rule). Create a worktree for the block repo under `worktrees/sequence-properties//` (use the `worktree` skill). Ask the operator whether this is tied to a Notion ticket (`MILAB-NNNN`); if yes, name the branch `MILAB-NNNN_seqprops-parallel-compute` and prefix every commit subject `MILAB-NNNN: …`. If no ticket, use `feat/seqprops-parallel-compute`. +2. **Move this plan into the worktree** and commit it as the first commit on the branch, so it travels with the work. +3. **Baseline green run** — from `blocks/sequence-properties/software/`: + ```bash + uv sync --locked + uv run pytest + uv run ruff check + uv run ruff format --check + ``` + All must pass before you change anything. If they don't, stop and tell the operator — you cannot characterize behavior on a red baseline. +4. **All Python commands run from `blocks/sequence-properties/software/`** (the dir holding `pyproject.toml`). All `pnpm` build commands run from the block root `blocks/sequence-properties/`. + +## The Determinism Contract (the invariant every task must preserve) + +The block's output must be **byte-identical run-to-run and machine-to-machine** so the platform's content-addressed cache dedups it. The current code earns this with two mechanisms — do not break either: + +- **Stable row order:** output rows are sorted by key before writing (`io_layer.py:48-50`, `main.py:47-48`). +- **FP reproducibility:** `charge_*`, `chargeShift_*`, `pi_*` touch the `10^x` transcendental and are rounded to 3 decimals at the output boundary (`pipeline.py:55-63`); every other property is bit-exact integer/constant arithmetic given a **fixed summation order**. + +The governing rule for this plan: **parallelism across rows is safe** (each row is an independent, pure computation; results are reassembled in input order; the final sort restores byte order). **Parallelism inside a single float reduction is not** (it reorders a sum). Phase B parallelizes only across rows — it never changes how any single property is summed. The golden master in Phase A is what proves this held. + +## File Structure + +| Path | Responsibility | Touched in | +|---|---|---| +| `software/tests/regression/__init__.py` | New regression-suite package marker | Task 1 | +| `software/tests/regression/golden_cases.py` | Single source of truth for golden input rows + plans (DRY: shared by regen + test) | Task 1 | +| `software/tests/regression/regen_golden.py` | Deliberate regeneration of committed golden outputs from current code | Task 1 | +| `software/tests/regression/test_golden_master.py` | Asserts full-output bytes (sha256) match committed goldens | Task 1 | +| `software/tests/data/golden//{properties.tsv,aa_fraction.tsv,stats.json}` | Committed frozen outputs (generated artifacts) | Task 1 | +| `software/tests/regression/test_scale_invariants.py` | Structural invariants at scale: count/keys preserved, byte-stable | Task 2 | +| `software/src/pipeline.py` | Add worker pool + `resolve_workers`; thread `workers` through `run` | Task 3 | +| `software/tests/unit/test_parallel_invariance.py` | Worker-count invariance: `workers=1` ≡ `workers=N`, byte-identical | Task 3 | +| `software/src/main.py` | Resolve worker count from env (argv unchanged), pass to `run`, log it | Task 4 | +| `workflow/src/main.tpl.tengo` | Request more cores/mem on the Python exec step | Task 5 | +| `software/.changeset/*.md`, `workflow/.changeset` (root `.changeset/`) | Version bump for software + workflow | Task 5 | +| `software/src/io_layer.py`, `software/src/pipeline.py` | (OPTIONAL) streaming sorted-batch I/O | Task 6 | + +--- + +## Phase A — Characterization (do first) + +### Task 1: Freeze a full-output golden master + +**Goal:** A committed, byte-exact snapshot of the full CLI output (`properties.tsv`, `aa_fraction.tsv`, `stats.json`) for every mode + coverage tier + edge case, generated from the *current* (pre-refactor) code, asserted via sha256. This is the safety net for Phases B and C. + +**Files:** +- Create: `software/tests/regression/__init__.py` +- Create: `software/tests/regression/golden_cases.py` +- Create: `software/tests/regression/regen_golden.py` +- Create: `software/tests/regression/test_golden_master.py` +- Create (generated): `software/tests/data/golden//{properties.tsv,aa_fraction.tsv,stats.json}` + +**Acceptance Criteria:** +- [ ] Cases cover: peptide (valid, <10aa→II NA, no-aromatic ε=0, paired-Cys ε, non-standard residue excluded, all-non-standard→all NA, stop-codon→all NA, empty→all NA), antibody full+Fv (incl. a clone with one region missing → full-chain NA for that clone, and a clone with empty CDR3), antibody CDR3-only, antibody partial (`fullChains:["A"]`), TCR (no Fv columns). +- [ ] `test_golden_master.py` runs `main()` per case into a tmp dir and asserts sha256 of all three output files equals the committed golden. +- [ ] A non-vacuous guard: at least one case asserts a known column set is present (proves the snapshot isn't empty). +- [ ] Goldens are generated by `regen_golden.py` from current code and committed. + +**Verify:** `uv run pytest tests/regression/test_golden_master.py -v` → all cases PASS. + +**Steps:** + +- [ ] **Step 1: Create the regression package marker** + +Create `software/tests/regression/__init__.py` (empty file). + +- [ ] **Step 2: Define the golden cases (single source of truth)** + +Create `software/tests/regression/golden_cases.py`: + +```python +"""Golden-master case definitions — the single source of truth for both the +regeneration script and the byte-compare test. Each case is (input rows, +column order, plan dict). Inputs live in code; only the *outputs* are +committed under tests/data/golden//. + +Edge cases are drawn from the spec's documented behaviour tables (NA +propagation, non-standard residues, stop codon, per-clone missing region, +TCR has no Fv). +""" + +from __future__ import annotations + +_AB_COLS = ( + ["entity_key"] + + [f"A_{f}" for f in ("FR1", "CDR1", "FR2", "CDR2", "FR3", "CDR3", "FR4")] + + [f"B_{f}" for f in ("FR1", "CDR1", "FR2", "CDR2", "FR3", "CDR3", "FR4")] +) + +_AB_FULL_ROW = { + "entity_key": "c1", + "A_FR1": "EVQLVES", "A_CDR1": "GFTFSSY", "A_FR2": "AMSWVRQ", "A_CDR2": "ISGSGGS", + "A_FR3": "TYYAESVKGRFTI", "A_CDR3": "CARDYW", "A_FR4": "WGQGTLV", + "B_FR1": "DIQMTQS", "B_CDR1": "QSISSY", "B_FR2": "LNWYQQK", "B_CDR2": "AASSLQS", + "B_FR3": "GVPSRFSGSG", "B_CDR3": "CQQYNS", "B_FR4": "FGQGTKV", +} +# Clone with one heavy region missing → full-chain A is NA for this clone only. +_AB_MISSING_REGION = {**_AB_FULL_ROW, "entity_key": "c2", "A_FR3": "", "A_CDR3": "CARGFW", "B_CDR3": "CQHFSS"} +# Clone with empty heavy CDR3 → CDR3-A NA for this clone only. +_AB_EMPTY_CDR3 = {**_AB_FULL_ROW, "entity_key": "c3", "A_CDR3": ""} + +# CASES: name -> (rows, columns, plan) +CASES: dict[str, tuple[list[dict], list[str], dict]] = { + "peptide": ( + [ + {"entity_key": "p_valid", "sequence": "ACDEFGHIKL"}, + {"entity_key": "p_basic", "sequence": "KKKKHHHHHH"}, + {"entity_key": "p_acidic", "sequence": "DDDDEEEEEE"}, + {"entity_key": "p_short", "sequence": "RPPGFSPF"}, # 8 aa -> instability NA + {"entity_key": "p_no_aromatic", "sequence": "AAAAAAAAAA"}, # eox/ered = 0 + {"entity_key": "p_paired_cys", "sequence": "CYIQNCPLG"}, # ε floor(C/2)*125 + {"entity_key": "p_nonstd", "sequence": "ACDXEFGHIK"}, # X excluded + {"entity_key": "p_all_nonstd", "sequence": "XXXXX"}, # all NA + {"entity_key": "p_stop", "sequence": "ACDE*GHIK"}, # stop codon -> all NA + {"entity_key": "p_empty", "sequence": ""}, # empty -> all NA + ], + ["entity_key", "sequence"], + {"mode": "peptide"}, + ), + "antibody_full": ( + [_AB_FULL_ROW, _AB_MISSING_REGION, _AB_EMPTY_CDR3], + _AB_COLS, + {"mode": "antibody_tcr_legacy_bulk", "receptor": "IG", + "chains": ["A", "B"], "fullChains": ["A", "B"], "hasFv": True}, + ), + "antibody_cdr3_only": ( + [ + {"entity_key": "c1", "A_CDR3": "CARDYW", "B_CDR3": "CQQYNS"}, + {"entity_key": "c2", "A_CDR3": "CARGFW", "B_CDR3": "CQHFSS"}, + ], + ["entity_key", "A_CDR3", "B_CDR3"], + {"mode": "antibody_tcr_legacy_sc", "receptor": "IG", + "chains": ["A", "B"], "fullChains": [], "hasFv": False}, + ), + "antibody_partial": ( + [_AB_FULL_ROW], + _AB_COLS, + {"mode": "antibody_tcr_legacy_bulk", "receptor": "IG", + "chains": ["A", "B"], "fullChains": ["A"], "hasFv": False}, + ), + "tcr": ( + [_AB_FULL_ROW], + _AB_COLS, + {"mode": "antibody_tcr_legacy_bulk", "receptor": "TCRAB", + "chains": ["A", "B"], "fullChains": ["A", "B"], "hasFv": False}, + ), +} +``` + +- [ ] **Step 3: Write the regeneration helper** + +Create `software/tests/regression/regen_golden.py`: + +```python +"""Regenerate committed golden outputs from the CURRENT code. + +Run ONLY deliberately: + uv run python tests/regression/regen_golden.py + +DO NOT run this during the parallelism/streaming refactor — if the golden +needs regenerating to make test_golden_master pass, the refactor changed +observable behaviour and that is a bug, not a snapshot update. Regenerate +only on an intentional behaviour change (BioPython bump, pKa update, spec +change) and call it out in the PR. +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "src")) + +from main import main # noqa: E402 + +from golden_cases import CASES # noqa: E402 + +GOLDEN_DIR = Path(__file__).resolve().parents[1] / "data" / "golden" + + +def _write_tsv(path: Path, rows: list[dict], columns: list[str]) -> None: + lines = ["\t".join(columns)] + for row in rows: + lines.append("\t".join(row.get(c, "") for c in columns)) + path.write_text("\n".join(lines) + "\n") + + +def regen() -> None: + for name, (rows, columns, plan) in CASES.items(): + case_dir = GOLDEN_DIR / name + case_dir.mkdir(parents=True, exist_ok=True) + in_tsv = case_dir / "input.tsv" + plan_json = case_dir / "plan.json" + _write_tsv(in_tsv, rows, columns) + plan_json.write_text(json.dumps(plan)) + rc = main([ + "--input", str(in_tsv), + "--plan", str(plan_json), + "--output", str(case_dir / "properties.tsv"), + "--aa-fraction", str(case_dir / "aa_fraction.tsv"), + "--stats", str(case_dir / "stats.json"), + ]) + assert rc == 0, f"regen failed for case {name}" + print(f"regenerated golden/{name}") + + +if __name__ == "__main__": + regen() +``` + +- [ ] **Step 4: Write the byte-compare test (red — goldens don't exist yet)** + +Create `software/tests/regression/test_golden_master.py`: + +```python +"""Full-output byte snapshot — the refactor safety net. + +Each case runs main() into a tmp dir and asserts every output file is +byte-identical (sha256) to the committed golden produced by regen_golden.py +from pre-refactor code. If any of Phase B / Phase C changes a byte, this +fails — which is the entire point. +""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +import pytest + +from main import main +from golden_cases import CASES + +GOLDEN_DIR = Path(__file__).resolve().parents[1] / "data" / "golden" +_OUTPUTS = ("properties.tsv", "aa_fraction.tsv", "stats.json") + + +def _sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _write_tsv(path: Path, rows: list[dict], columns: list[str]) -> None: + lines = ["\t".join(columns)] + for row in rows: + lines.append("\t".join(row.get(c, "") for c in columns)) + path.write_text("\n".join(lines) + "\n") + + +@pytest.mark.parametrize("name", sorted(CASES), ids=sorted(CASES)) +def test_output_matches_golden(tmp_path: Path, name: str): + rows, columns, plan = CASES[name] + in_tsv = tmp_path / "input.tsv" + plan_json = tmp_path / "plan.json" + _write_tsv(in_tsv, rows, columns) + plan_json.write_text(json.dumps(plan)) + + out = {n: tmp_path / n for n in _OUTPUTS} + rc = main([ + "--input", str(in_tsv), + "--plan", str(plan_json), + "--output", str(out["properties.tsv"]), + "--aa-fraction", str(out["aa_fraction.tsv"]), + "--stats", str(out["stats.json"]), + ]) + assert rc == 0 + + golden = GOLDEN_DIR / name + assert golden.is_dir(), f"missing golden dir for {name}; run regen_golden.py" + for n in _OUTPUTS: + assert _sha256(out[n]) == _sha256(golden / n), ( + f"{name}/{n} diverged from golden — refactor changed behaviour" + ) + + +# Non-vacuous guard: the antibody_full golden actually carries the Fv columns, +# and the tcr golden does NOT (R12). Proves the snapshot is meaningful, not empty. +def test_golden_column_presence(): + import polars as pl + + ab = pl.read_csv(GOLDEN_DIR / "antibody_full" / "properties.tsv", separator="\t") + assert {"charge_Fv", "pi_Fv", "charge_A_VDJRegion"} <= set(ab.columns) + tcr = pl.read_csv(GOLDEN_DIR / "tcr" / "properties.tsv", separator="\t") + assert not any(c.endswith("_Fv") for c in tcr.columns) +``` + +- [ ] **Step 5: Run the test — confirm it fails for the right reason** + +Run: `uv run pytest tests/regression/test_golden_master.py -v` +Expected: FAIL — `missing golden dir for ; run regen_golden.py`. + +- [ ] **Step 6: Generate goldens from current code, then go green** + +```bash +uv run python tests/regression/regen_golden.py +uv run pytest tests/regression/test_golden_master.py -v +``` +Expected: regen prints one line per case; pytest PASSES all cases + `test_golden_column_presence`. + +- [ ] **Step 7: Lint, then commit (goldens included)** + +```bash +uv run ruff format +uv run ruff check +git add tests/regression/ tests/data/golden/ +git commit -m "test: freeze full-output golden master for sequence-properties refactor" +``` + +--- + +### Task 2: Structural invariants at scale + +**Goal:** Pin the behavioral invariants that the parallel/streaming refactor must preserve — output has exactly one row per input entity (no loss/duplication), the same set of keys, the AA-fraction frame has 20 rows per peptide, and two in-process runs are byte-identical — at a size (5000 rows) that will span multiple worker chunks in Phase B. + +**Files:** +- Create: `software/tests/regression/test_scale_invariants.py` + +**Acceptance Criteria:** +- [ ] Deterministic synthetic input (seeded `random.Random(0)` — no unseeded RNG, no clock). +- [ ] Peptide: output row count == input count; output `entity_key` set == input set; AA-fraction height == 20 × count. +- [ ] Antibody full: output row count == input count; key set preserved; `charge_Fv` column present. +- [ ] Two runs on the same input produce byte-identical `properties.tsv`. +- [ ] Marked `@pytest.mark.slow` so it deselects during fast iteration. + +**Verify:** `uv run pytest tests/regression/test_scale_invariants.py -v` → PASS (and `-m "not slow"` deselects it). + +**Steps:** + +- [ ] **Step 1: Write the invariant test (passes on current sequential code)** + +Create `software/tests/regression/test_scale_invariants.py`: + +```python +"""Structural invariants at scale — the per-row-map contract the parallel and +streaming refactors must preserve: one output row per input entity, same key +set, AA-fraction 20 rows/peptide, byte-stable across runs. + +Synthetic data is built from a SEEDED RNG (random.Random(0)) so the suite is +deterministic. Marked slow: 5000 rows is enough to exercise >1 worker chunk +once Phase B lands, but too slow for every iteration. +""" + +from __future__ import annotations + +import hashlib +import random +from pathlib import Path + +import polars as pl +import pytest + +from io_layer import write_output_tsv +from pipeline import run + +_AAS = "ACDEFGHIKLMNPQRSTVWY" +_REGIONS = ("FR1", "CDR1", "FR2", "CDR2", "FR3", "CDR3", "FR4") + + +def _rand_seq(rng: random.Random, lo: int, hi: int) -> str: + return "".join(rng.choice(_AAS) for _ in range(rng.randint(lo, hi))) + + +def _peptide_frame(n: int) -> pl.DataFrame: + rng = random.Random(0) + return pl.DataFrame( + {"entity_key": [f"p{i}" for i in range(n)], + "sequence": [_rand_seq(rng, 5, 25) for _ in range(n)]}, + schema={"entity_key": pl.Utf8, "sequence": pl.Utf8}, + ) + + +def _antibody_frame(n: int) -> pl.DataFrame: + rng = random.Random(1) + cols: dict[str, list[str]] = {"entity_key": [f"c{i}" for i in range(n)]} + for ch in ("A", "B"): + for feat in _REGIONS: + cols[f"{ch}_{feat}"] = [_rand_seq(rng, 6, 14) for _ in range(n)] + schema = {k: pl.Utf8 for k in cols} + return pl.DataFrame(cols, schema=schema) + + +_AB_PLAN = {"mode": "antibody_tcr_legacy_bulk", "receptor": "IG", + "chains": ["A", "B"], "fullChains": ["A", "B"], "hasFv": True} + +N = 5000 + + +@pytest.mark.slow +def test_peptide_row_and_key_preservation(): + reads = _peptide_frame(N) + out = run(reads, {"mode": "peptide"}) + props = out["properties"] + assert props.height == N + assert set(props["entity_key"].to_list()) == set(reads["entity_key"].to_list()) + assert out["aa_fraction"].height == 20 * N + + +@pytest.mark.slow +def test_antibody_row_and_key_preservation(): + reads = _antibody_frame(N) + out = run(reads, _AB_PLAN) + props = out["properties"] + assert props.height == N + assert set(props["entity_key"].to_list()) == set(reads["entity_key"].to_list()) + assert "charge_Fv" in props.columns + + +@pytest.mark.slow +def test_peptide_byte_stable_two_runs(tmp_path: Path): + reads = _peptide_frame(N) + a, b = tmp_path / "a.tsv", tmp_path / "b.tsv" + write_output_tsv(run(reads, {"mode": "peptide"})["properties"], a, sort_keys=["entity_key"]) + write_output_tsv(run(reads, {"mode": "peptide"})["properties"], b, sort_keys=["entity_key"]) + assert hashlib.sha256(a.read_bytes()).hexdigest() == hashlib.sha256(b.read_bytes()).hexdigest() +``` + +- [ ] **Step 2: Run it (and confirm slow-deselect works)** + +```bash +uv run pytest tests/regression/test_scale_invariants.py -v +uv run pytest tests/regression/test_scale_invariants.py -m "not slow" # collects 0 +``` +Expected: first PASSES (3 tests); second deselects all 3. + +- [ ] **Step 3: Lint and commit** + +```bash +uv run ruff format +uv run ruff check +git add tests/regression/test_scale_invariants.py +git commit -m "test: pin row/key-preservation + byte-stability invariants at scale" +``` + +--- + +## Phase B — Option #1: Parallelize across cores + +### Task 3: Map the per-row compute across a process pool + +**Goal:** Run the existing pure per-row compute functions across a `ProcessPoolExecutor`, assembling results in input order, so the output is byte-identical regardless of worker count. Keep a sequential fallback for small inputs and `workers=1`. + +**Files:** +- Modify: `software/src/pipeline.py` (add imports, `resolve_workers`, `_pmap`, worker functions; thread `workers` through `run` / `run_peptide` / `run_antibody_tcr` — `pipeline.py:71-91`, `159-221`, `414-439`) +- Test: `software/tests/unit/test_parallel_invariance.py` (create) + +**Acceptance Criteria:** +- [ ] `run(reads, plan, workers=None)` accepts a worker count; `run_peptide` / `run_antibody_tcr` accept and use it. +- [ ] `workers=1` and `workers=4` produce byte-identical `properties.tsv`, `aa_fraction.tsv`, and identical `stats` for peptide, antibody-full, and the 5000-row scale frame. +- [ ] Worker functions are module-level and picklable (spawn-safe): they take only strings / plain dicts. +- [ ] Results are assembled in input order (no reliance on completion order, no `set`/`dict`-order leakage into output). +- [ ] The Phase A golden master and scale invariants still pass unchanged. + +**Verify:** `uv run pytest tests/unit/test_parallel_invariance.py tests/regression/ -v` → PASS. + +**Steps:** + +- [ ] **Step 1: Write the invariance test first (red)** + +Create `software/tests/unit/test_parallel_invariance.py`: + +```python +"""Worker-count invariance — the core safety property for parallelism. + +The output of run() must not depend on how many workers compute it. We +compare workers=1 against workers=4 (and a larger frame that actually spills +into the pool) and assert byte-identical serialized output + identical stats. +""" + +from __future__ import annotations + +import random +from pathlib import Path + +import polars as pl +import pytest + +from io_layer import write_output_tsv +from pipeline import run + +_AAS = "ACDEFGHIKLMNPQRSTVWY" +_REGIONS = ("FR1", "CDR1", "FR2", "CDR2", "FR3", "CDR3", "FR4") +_AB_PLAN = {"mode": "antibody_tcr_legacy_bulk", "receptor": "IG", + "chains": ["A", "B"], "fullChains": ["A", "B"], "hasFv": True} + + +def _peptide_frame(n: int, seed: int) -> pl.DataFrame: + rng = random.Random(seed) + seqs = ["".join(rng.choice(_AAS) for _ in range(rng.randint(5, 25))) for _ in range(n)] + return pl.DataFrame({"entity_key": [f"p{i}" for i in range(n)], "sequence": seqs}, + schema={"entity_key": pl.Utf8, "sequence": pl.Utf8}) + + +def _antibody_frame(n: int, seed: int) -> pl.DataFrame: + rng = random.Random(seed) + cols = {"entity_key": [f"c{i}" for i in range(n)]} + for ch in ("A", "B"): + for feat in _REGIONS: + cols[f"{ch}_{feat}"] = ["".join(rng.choice(_AAS) for _ in range(rng.randint(6, 14))) for _ in range(n)] + return pl.DataFrame(cols, schema={k: pl.Utf8 for k in cols}) + + +def _serialize(out: dict, tmp: Path, tag: str) -> tuple[bytes, bytes, dict]: + p = tmp / f"props_{tag}.tsv" + a = tmp / f"aa_{tag}.tsv" + write_output_tsv(out["properties"], p, sort_keys=["entity_key"]) + write_output_tsv(out["aa_fraction"], a, sort_keys=["entity_key", "aminoAcid"]) + return p.read_bytes(), a.read_bytes(), out["stats"] + + +@pytest.mark.parametrize("n", [3, 3000], ids=["sequential-path", "pool-path"]) +def test_peptide_invariant_to_workers(tmp_path: Path, n: int): + reads = _peptide_frame(n, seed=0) + one = _serialize(run(reads, {"mode": "peptide"}, workers=1), tmp_path, "1") + four = _serialize(run(reads, {"mode": "peptide"}, workers=4), tmp_path, "4") + assert one == four + + +@pytest.mark.parametrize("n", [3, 3000], ids=["sequential-path", "pool-path"]) +def test_antibody_invariant_to_workers(tmp_path: Path, n: int): + reads = _antibody_frame(n, seed=1) + one = _serialize(run(reads, _AB_PLAN, workers=1), tmp_path, "1") + four = _serialize(run(reads, _AB_PLAN, workers=4), tmp_path, "4") + assert one == four +``` + +- [ ] **Step 2: Run it — confirm it fails (run() has no `workers` param yet)** + +Run: `uv run pytest tests/unit/test_parallel_invariance.py -v` +Expected: FAIL — `TypeError: run() got an unexpected keyword argument 'workers'`. + +- [ ] **Step 3: Add the pool infrastructure to `pipeline.py`** + +At the top of `software/src/pipeline.py`, add to the imports block (after `import logging`): + +```python +import functools +import os +from concurrent.futures import ProcessPoolExecutor +``` + +Then add, just below the `PH = 7.0` line (`pipeline.py:34`): + +```python +# Parallelism. Below this row count the pool's process startup + pickle cost +# outweighs the benefit, so we stay in-process. The threshold also keeps the +# whole existing unit-test suite on the fast sequential path. +_PARALLEL_MIN_ROWS = 2000 + + +def resolve_workers(workers: int | None) -> int: + """How many worker processes to use. Explicit arg wins (used by tests and + by main.py once it reads the platform's CPU allocation). Falls back to the + PL_COMPUTE_WORKERS env var, then os.cpu_count(). The RESULT never depends on + this number — only the wall-clock does — so an over- or under-estimate is a + speed concern, never a correctness one. + """ + if workers is not None: + return max(1, int(workers)) + env = os.environ.get("PL_COMPUTE_WORKERS") + if env and env.isdigit() and int(env) > 0: + return int(env) + return max(1, os.cpu_count() or 1) + + +def _pmap(fn, items: list, workers: int, chunksize: int = 256) -> list: + """Map fn over items, preserving input order. Sequential below the + threshold or when workers<=1; otherwise a process pool. ProcessPoolExecutor + .map() preserves input order, so results align with items by index — the + property the byte-stable output depends on. + """ + if workers <= 1 or len(items) < _PARALLEL_MIN_ROWS: + return [fn(x) for x in items] + with ProcessPoolExecutor(max_workers=workers) as ex: + return list(ex.map(fn, items, chunksize=chunksize)) +``` + +- [ ] **Step 4: Add module-level, picklable worker functions** + +In `software/src/pipeline.py`, add a peptide worker just above `run_peptide` (`pipeline.py:159`): + +```python +def _peptide_worker(seq: str) -> tuple[dict[str, float | None], list[float | None] | None]: + """Picklable per-peptide unit: (properties row, 20 AA fractions in + STANDARD_AAS order | None). One SequenceContext per sequence, shared across + all 11 reads — same sharing the old inline loop relied on. + """ + ctx = SequenceContext.from_seq(seq) + if ctx is None: + return (dict(_NA_PEPTIDE_ROW), None) + props = _compute_peptide_row_from_ctx(ctx) + fr = ctx.aa_fractions() + return (props, [fr[aa] for aa in STANDARD_AAS]) +``` + +And an antibody worker just above `run_antibody_tcr` (`pipeline.py:414`): + +```python +def _antibody_worker(record: dict, plan: dict[str, Any]) -> dict[str, Any]: + """Picklable per-clone unit. `plan` is bound per-call via functools.partial; + it is a plain dict and pickles cleanly. Delegates to the existing + _compute_row_for so the computation is unchanged. + """ + return _compute_row_for(record, plan) +``` + +- [ ] **Step 5: Rewrite `run` to thread `workers` through** + +Replace `run` (`pipeline.py:71-91`) with: + +```python +def run(reads: pl.DataFrame, plan: dict[str, Any], workers: int | None = None) -> dict[str, Any]: + """Dispatch by mode. `workers` controls parallelism only — output is + identical for any value (see test_parallel_invariance). Returns a dict with + `properties`, `aa_fraction`, and `stats` entries (unchanged contract). + """ + n_workers = resolve_workers(workers) + mode = plan["mode"] + if mode == "peptide": + log.info("Running peptide mode (%d entities, %d workers)", reads.height, n_workers) + return run_peptide(reads, n_workers) + log.info( + "Running antibody/TCR mode (receptor=%s, %d clones, %d workers)", + plan.get("receptor", "IG"), reads.height, n_workers, + ) + return run_antibody_tcr(reads, plan, n_workers) +``` + +- [ ] **Step 6: Rewrite `run_peptide`'s loop to use the pool** + +Replace the body of `run_peptide` (`pipeline.py:159-221`) — keep the signature change and the DataFrame/stats construction identical, only the per-row work moves into `_pmap`: + +```python +def run_peptide(reads: pl.DataFrame, workers: int = 1) -> dict[str, Any]: + """Compute peptide-mode outputs. Per-sequence work runs through `_pmap` + (sequential or pooled); results are reassembled in input order so the + serialized output is byte-identical regardless of worker count. + """ + keys = reads["entity_key"].to_list() + seqs = reads["sequence"].to_list() + n = len(seqs) + + log.info("Computing peptide properties + AA fractions (%d sequences)", n) + results = _pmap(_peptide_worker, seqs, workers) + + prop_cols: dict[str, list[Any]] = {"entity_key": list(keys), **{c: [] for c in PEPTIDE_PROPERTY_COLUMNS}} + aa_entity: list[str] = [] + aa_amino: list[str] = [] + aa_value: list[float | None] = [] + for k, (props, fractions) in zip(keys, results): + for c in PEPTIDE_PROPERTY_COLUMNS: + prop_cols[c].append(props[c]) + if fractions is None: + for aa in STANDARD_AAS: + aa_entity.append(k) + aa_amino.append(aa) + aa_value.append(None) + else: + for aa, val in zip(STANDARD_AAS, fractions): + aa_entity.append(k) + aa_amino.append(aa) + aa_value.append(val) + properties = pl.DataFrame( + prop_cols, + schema={"entity_key": pl.Utf8, **{c: pl.Float64 for c in PEPTIDE_PROPERTY_COLUMNS}}, + ) + aa_fraction = pl.DataFrame( + {"entity_key": aa_entity, "aminoAcid": aa_amino, "value": aa_value}, + schema={"entity_key": pl.Utf8, "aminoAcid": pl.Utf8, "value": pl.Float64}, + ) + + has_below_floor = any(0 < effective_length(s) < INSTABILITY_MIN_LENGTH for s in seqs if s) + stats = { + "medianCdr3Length": {}, + "hasPeptideBelowInstabilityFloor": has_below_floor, + } + + return { + "properties": _quantize_for_cid(properties), + "aa_fraction": aa_fraction, + "stats": stats, + } +``` + +- [ ] **Step 7: Rewrite `run_antibody_tcr`'s loop to use the pool** + +Replace the body of `run_antibody_tcr` (`pipeline.py:414-439`): + +```python +def run_antibody_tcr(reads: pl.DataFrame, plan: dict[str, Any], workers: int = 1) -> dict[str, Any]: + chains = plan.get("chains", []) + full_chains = plan.get("fullChains", []) + n = reads.height + if chains: + log.info("Computing CDR3 properties for chains %s (%d clones)", list(chains), n) + if full_chains: + log.info("Reconstructing full chains %s and computing full-chain properties", list(full_chains)) + if plan.get("hasFv"): + log.info("Computing Fv properties (paired VH+VL)") + + out_cols = _planned_output_columns(plan) + records = list(reads.iter_rows(named=True)) + worker = functools.partial(_antibody_worker, plan=plan) + rows = _pmap(worker, records, workers) + + columns: dict[str, list[Any]] = {"entity_key": [], **{c: [] for c in out_cols}} + for row in rows: + columns["entity_key"].append(row["entity_key"]) + for c in out_cols: + columns[c].append(row[c]) + schema = {"entity_key": pl.Utf8, **{c: pl.Float64 for c in out_cols}} + properties = pl.DataFrame(columns, schema=schema) + aa_fraction = pl.DataFrame(schema={"entity_key": pl.Utf8, "aminoAcid": pl.Utf8, "value": pl.Float64}) + stats = {"medianCdr3Length": _median_cdr3_length_by_chain(reads, chains)} + return { + "properties": _quantize_for_cid(properties), + "aa_fraction": aa_fraction, + "stats": stats, + } +``` + +- [ ] **Step 8: Run the invariance test + the full safety net** + +```bash +uv run pytest tests/unit/test_parallel_invariance.py -v +uv run pytest tests/regression/ -v +uv run pytest +``` +Expected: invariance PASSES (both sequential-path and pool-path ids); golden master + scale invariants PASS unchanged; full suite green. If the golden master fails, **do not regenerate it** — the parallel path changed a byte; debug ordering/assembly until it matches. + +- [ ] **Step 9: Lint and commit** + +```bash +uv run ruff format +uv run ruff check +git add src/pipeline.py tests/unit/test_parallel_invariance.py +git commit -m "feat: parallelize per-row property compute across a process pool" +``` + +--- + +### Task 4: Resolve worker count from the environment in `main.py` + +**Goal:** Make the CLI use all allocated cores at runtime without adding anything to argv — so the workflow exec node's command line stays constant and its content-addressed identity does not change. + +**Files:** +- Modify: `software/src/main.py` (`main.py:18`, `main.py:44-46`) +- Test: `software/tests/integration/test_cli.py` (add one test) + +**Acceptance Criteria:** +- [ ] `main()` resolves workers via `pipeline.resolve_workers(None)` (env → `os.cpu_count()`), passes to `run`, and logs the count to stderr. +- [ ] **No new CLI argument** — argv is unchanged from today (preserves the exec CID). +- [ ] Output bytes are identical with `PL_COMPUTE_WORKERS=1` vs unset (env changes speed, not output). + +**Verify:** `uv run pytest tests/integration/test_cli.py -v` → PASS. + +**Steps:** + +- [ ] **Step 1: Add the env-invariance test (red)** + +Append to `software/tests/integration/test_cli.py`: + +```python +# Worker count comes from the environment, not argv, and must not change output +# bytes — only wall-clock. Guards the CID-stability property at the CLI layer. +def test_env_worker_count_does_not_change_output(tmp_path: Path, monkeypatch): + rows = [ + {"entity_key": "p1", "sequence": "ACDEFGHIKL"}, + {"entity_key": "p2", "sequence": "MNPQRSTVWY"}, + ] + monkeypatch.setenv("PL_COMPUTE_WORKERS", "1") + out_a, _ = _run_peptide(tmp_path, "_w1", rows) + monkeypatch.setenv("PL_COMPUTE_WORKERS", "4") + out_b, _ = _run_peptide(tmp_path, "_w4", rows) + assert _sha256(out_a) == _sha256(out_b) +``` + +- [ ] **Step 2: Run it — confirm it passes already (worker count not yet wired, but defaults are equivalent on tiny input)** + +Run: `uv run pytest tests/integration/test_cli.py::test_env_worker_count_does_not_change_output -v` +Expected: PASS (tiny input stays sequential either way). This test locks the contract; the wiring below makes it meaningful at scale. + +- [ ] **Step 3: Wire `resolve_workers` into `main.py`** + +In `software/src/main.py`, change the import (`main.py:19`): + +```python +from pipeline import resolve_workers, run +``` + +Then in `main()`, replace the `outputs = run(reads, plan)` line (`main.py:46`) with: + +```python + workers = resolve_workers(None) + logging.getLogger(__name__).info("Using %d compute worker(s)", workers) + outputs = run(reads, plan, workers=workers) +``` + +- [ ] **Step 4: Run the CLI suite + full suite** + +```bash +uv run pytest tests/integration/test_cli.py -v +uv run pytest +``` +Expected: all PASS, including the determinism subprocess test (`test_determinism.py`) which now also exercises env-resolved workers in a fresh process. + +- [ ] **Step 5: Lint and commit** + +```bash +uv run ruff format +uv run ruff check +git add src/main.py tests/integration/test_cli.py +git commit -m "feat: resolve compute worker count from environment in CLI" +``` + +--- + +### Task 5: Request more cores + memory on the workflow exec step + +**Goal:** Tell the platform to schedule the Python step with multiple cores and more memory, so the parallel kernel actually has cores to use and a large input doesn't OOM. Keep the exec node's identity stable. + +**Files:** +- Modify: `workflow/src/main.tpl.tengo` (`main.tpl.tengo:344-360`) +- Create: `.changeset/.md` (block root) +- Verify against SDK: `core/platforma/sdk/workflow-tengo/src/exec/index.lib.tengo` + +**Acceptance Criteria:** +- [ ] The `pyRun` exec builder requests multiple cores (`.cpu(8)`) and more memory (`.mem("8GiB")`). +- [ ] **No `--workers` (or any machine-dependent value) added to argv** — Python reads cores from the environment at runtime. +- [ ] Confirmed (by reading `exec/index.lib.tengo`) whether `cpu()`/`mem()` enter the exec node's content-addressed id. Document the finding in the PR description: if they do, the bump is a one-time cache invalidation and the requested values are fixed constants in code, so the CID stays stable run-to-run. +- [ ] `pnpm run build:dev` succeeds. +- [ ] A changeset covers `.workflow` and `.software`. + +**Verify:** `pnpm run build:dev` (from block root) → build succeeds; `grep -n "cpu(" workflow/src/main.tpl.tengo` shows the new value on the `pyRun` step. + +**Steps:** + +- [ ] **Step 1: Verify whether cpu()/mem() are part of the exec CID** + +Read `core/platforma/sdk/workflow-tengo/src/exec/index.lib.tengo` and confirm whether `.cpu()` / `.mem()` feed the resource's content-addressed identity or are scheduling-only hints. Record the answer in your working notes — it determines whether bumping them invalidates existing cached results (a one-time, acceptable cost) and confirms the CID stays stable across machines (the requested values are fixed code constants, not machine-derived). + +- [ ] **Step 2: Bump cpu + mem on the Python step** + +In `workflow/src/main.tpl.tengo`, in the `pyRun := exec.builder()...` chain (`main.tpl.tengo:345-360`), change: + +```tengo + software(soft). + mem("4GiB"). + cpu(1). +``` +to: +```tengo + software(soft). + mem("8GiB"). + cpu(8). +``` + +Leave the `arg("--input")...arg("--stats")` chain exactly as-is — **do not add a `--workers` arg** (argv must stay constant for CID stability; the Python step reads its worker count from the environment via `resolve_workers`). + +- [ ] **Step 3: Add a changeset** + +Create `.changeset/seqprops-parallel-compute.md` (from block root): + +```markdown +--- +'@platforma-open/milaboratories.sequence-properties.software': patch +'@platforma-open/milaboratories.sequence-properties.workflow': patch +--- + +Parallelize the property computation across CPU cores and raise the compute step's CPU/memory request, so large datasets compute on the first run without the previous single-core bottleneck. Output is byte-identical to the prior version (worker count never affects results). +``` + +Do not hand-edit any `package.json` `version` field — changesets + CI handle the bump. + +- [ ] **Step 4: Build the block (dev)** + +From the block root `blocks/sequence-properties/`: +```bash +pnpm install +pnpm run build:dev +``` +Expected: turbo builds workflow + model + ui + software; build succeeds. If `pnpm install` modified `pnpm-lock.yaml`, stage it. + +- [ ] **Step 5: Commit** + +```bash +git add workflow/src/main.tpl.tengo .changeset/seqprops-parallel-compute.md +git add -A pnpm-lock.yaml 2>/dev/null || true +git commit -m "feat: request 8 cores + 8GiB for the sequence-properties compute step" +``` + +- [ ] **Step 6: (Operator / optional) live verification** + +The block-level workflow test (`workflow/src/wf.test.ts`) needs a live backend (`run-platforma` skill). If a backend is available, run the block test suite per the `block-dev` skill (`pnpm test --filter=...`). Otherwise the Python golden master + invariance tests are the primary correctness gate; flag to the operator that an end-to-end run on a representative large dataset should be done before marking the PR ready. + +--- + +## Phase C — Option #7: Streaming I/O (OPTIONAL — decide at end) + +> **OPTIONAL.** Do not implement until Phase B is merged and measured. Decision gate at Step 0 below. Phase B's `mem("8GiB")` bump may already make this unnecessary — streaming is only worth its complexity if a target dataset still risks OOM at the raised memory ceiling. + +### Task 6 (OPTIONAL): Stream sorted input in batches to bound peak memory + +**Goal:** Process the input in sorted batches and write output incrementally, so peak memory is O(batch) instead of O(dataset). Output stays byte-identical (the golden master is the guard). + +**Files:** +- Modify: `software/src/io_layer.py`, `software/src/pipeline.py` +- Test: `software/tests/regression/test_streaming_equivalence.py` (create) + +**Acceptance Criteria:** +- [ ] Decision recorded (see Step 0): implement or skip, with the measurement that justified it. +- [ ] If implemented: a streaming path that (a) sorts input by `entity_key` once up front so output needs no global re-sort, (b) computes + writes in batches of `B` rows (header on first batch only), (c) computes `stats` (median CDR3) without materializing the whole dataset. +- [ ] If implemented: `test_streaming_equivalence.py` asserts the streaming output is byte-identical to the non-streaming output on the 5000-row frame; the Phase A golden master still passes. + +**Steps:** + +- [ ] **Step 0: Decision gate (do this first, record the outcome in the PR)** + +Measure peak RSS of the Phase B build on the largest realistic input (or a synthetic frame at the agreed upper bound — confirm the target row count with the operator: 10⁶? 10⁷?). If peak memory stays comfortably under the `8GiB` request, **skip this task** and record "streaming not needed: peak RSS = X GiB at N rows < 8 GiB". Only proceed if it approaches or exceeds the ceiling. + +- [ ] **Step 1: Write the equivalence test (red)** + +Create `software/tests/regression/test_streaming_equivalence.py`: + +```python +"""Streaming must not change a byte. Compares the streaming CLI path against +the in-memory path on a 5000-row frame, and relies on the golden master for +the edge-case modes. +""" + +from __future__ import annotations + +import hashlib +import json +import random +from pathlib import Path + +import pytest + +from main import main + +_AAS = "ACDEFGHIKLMNPQRSTVWY" + + +def _write_peptides(path: Path, n: int) -> None: + rng = random.Random(0) + lines = ["entity_key\tsequence"] + for i in range(n): + seq = "".join(rng.choice(_AAS) for _ in range(rng.randint(5, 25))) + lines.append(f"p{i}\t{seq}") + path.write_text("\n".join(lines) + "\n") + + +def _run(tmp: Path, tag: str, in_tsv: Path, env_streaming: str, monkeypatch) -> str: + monkeypatch.setenv("PL_STREAM_BATCH", env_streaming) # "0" disables, ">0" sets batch size + plan = tmp / f"plan_{tag}.json" + plan.write_text(json.dumps({"mode": "peptide"})) + out = tmp / f"out_{tag}.tsv" + rc = main(["--input", str(in_tsv), "--plan", str(plan), "--output", str(out), + "--aa-fraction", str(tmp / f"aa_{tag}.tsv"), "--stats", str(tmp / f"stats_{tag}.json")]) + assert rc == 0 + return hashlib.sha256(out.read_bytes()).hexdigest() + + +@pytest.mark.slow +def test_streaming_matches_in_memory(tmp_path: Path, monkeypatch): + in_tsv = tmp_path / "input.tsv" + _write_peptides(in_tsv, 5000) + in_memory = _run(tmp_path, "mem", in_tsv, "0", monkeypatch) + streamed = _run(tmp_path, "stream", in_tsv, "1000", monkeypatch) + assert in_memory == streamed +``` + +- [ ] **Step 2: Implement the streaming path** + +In `software/src/pipeline.py`, add a streaming entry that the CLI selects when `PL_STREAM_BATCH` > 0: +- Sort `reads` by `entity_key` once (`reads.sort("entity_key")`). +- Iterate in slices of `B` rows; for each slice call the existing `run_peptide` / `run_antibody_tcr` (which already parallelize via Phase B), apply `_quantize_for_cid`, and append to the output TSV — header only on the first slice (`write_csv(..., include_header=(first_slice))`, append mode). +- Compute `stats` separately: `hasPeptideBelowInstabilityFloor` is a boolean OR across slices; `medianCdr3Length` is computed once over just the CDR3 columns (lengths are small ints — a single `effective_length` pass, no full compute). +- Because the input is pre-sorted, the appended output is already in `entity_key` order — no global re-sort, no full-dataset materialization. + +In `software/src/main.py`, read `int(os.environ.get("PL_STREAM_BATCH", "0"))` and dispatch to the streaming entry when > 0; otherwise the in-memory `run`. (Keep argv unchanged — streaming is env-gated, like worker count.) + +- [ ] **Step 3: Verify byte-equivalence + golden master** + +```bash +uv run pytest tests/regression/test_streaming_equivalence.py -v +uv run pytest tests/regression/test_golden_master.py -v +uv run pytest +``` +Expected: streaming output == in-memory output; golden master unchanged; full suite green. + +- [ ] **Step 4: Lint, changeset, commit** + +```bash +uv run ruff format +uv run ruff check +``` +Add a `patch` changeset entry for `.software`, then: +```bash +git add src/pipeline.py src/main.py tests/regression/test_streaming_equivalence.py .changeset/ +git commit -m "feat: optional streaming compute path to bound peak memory" +``` + +--- + +## Self-Review + +- **Spec coverage / scope:** This plan does not change any property formula, column, axis, domain, or annotation in the `sequence-properties` spec — it is a pure performance change behind a behavior-freeze. The determinism contract from `docs/text/work/projects/sequence-properties` (byte-stable output, CID quantization) is preserved by construction and verified by Task 1. +- **Characterization-first:** Phase A (Tasks 1–2) is committed before any compute change; the golden master is generated from pre-refactor code and must not be regenerated during Phases B/C. +- **Type/name consistency:** `resolve_workers`, `_pmap`, `_peptide_worker`, `_antibody_worker` are defined in Task 3 and consumed by Tasks 3–4; `run(reads, plan, workers=None)`, `run_peptide(reads, workers)`, `run_antibody_tcr(reads, plan, workers)` signatures are consistent across tasks; `PL_COMPUTE_WORKERS` (Task 3/4) and `PL_STREAM_BATCH` (Task 6) are the only new env knobs, both output-invariant. +- **CID stability:** every speed knob (worker count, streaming batch) is read from the environment at runtime, never added to argv — so the exec node's identity is unchanged. Task 5 Step 1 verifies the cpu/mem question against the SDK rather than assuming. +- **Open items to confirm with the operator at execution time:** the Notion ticket ID (branch naming), the realistic upper-bound row count (sizes the Task 6 decision gate), and the exact platform env var for allocated CPUs (Task 4 — `os.cpu_count()` is the safe default if none exists, since the result is worker-count-invariant). diff --git a/docs/superpowers/plans/2026-06-10-sequence-properties-parallel-compute.md.tasks.json b/docs/superpowers/plans/2026-06-10-sequence-properties-parallel-compute.md.tasks.json new file mode 100644 index 0000000..8bdf518 --- /dev/null +++ b/docs/superpowers/plans/2026-06-10-sequence-properties-parallel-compute.md.tasks.json @@ -0,0 +1,46 @@ +{ + "planPath": "docs/superpowers/plans/2026-06-10-sequence-properties-parallel-compute.md", + "tasks": [ + { + "id": 1, + "subject": "Task 1: Freeze full-output golden master", + "status": "pending", + "description": "**Goal:** Committed byte-exact snapshot of full CLI output for every mode + coverage tier + edge case, generated from current pre-refactor code, asserted via sha256. Safety net for Phases B/C.\n\n**Files:** software/tests/regression/{__init__.py, golden_cases.py, regen_golden.py, test_golden_master.py}; software/tests/data/golden//{properties.tsv,aa_fraction.tsv,stats.json}\n\n**Verify:** uv run pytest tests/regression/test_golden_master.py -v\n\n```json:metadata\n{\"files\": [\"software/tests/regression/__init__.py\",\"software/tests/regression/golden_cases.py\",\"software/tests/regression/regen_golden.py\",\"software/tests/regression/test_golden_master.py\",\"software/tests/data/golden/\"], \"verifyCommand\": \"uv run pytest tests/regression/test_golden_master.py -v\", \"acceptanceCriteria\": [\"all golden cases byte-match via sha256\",\"cases cover all modes/tiers/edge-cases\",\"goldens generated from pre-refactor code and committed\"]}\n```" + }, + { + "id": 2, + "subject": "Task 2: Structural invariants at scale", + "status": "pending", + "description": "**Goal:** Pin invariants the parallel/streaming refactor must preserve — one output row per input entity, same key set, AA-fraction 20 rows/peptide, two runs byte-identical — at 5000 rows.\n\n**Files:** software/tests/regression/test_scale_invariants.py\n\n**Verify:** uv run pytest tests/regression/test_scale_invariants.py -v\n\n```json:metadata\n{\"files\": [\"software/tests/regression/test_scale_invariants.py\"], \"verifyCommand\": \"uv run pytest tests/regression/test_scale_invariants.py -v\", \"acceptanceCriteria\": [\"row count + key set preserved\",\"aa_fraction 20 rows/peptide\",\"byte-stable across two runs\",\"seeded RNG, marked slow\"]}\n```" + }, + { + "id": 3, + "subject": "Task 3: Parallelize per-row compute across a process pool", + "status": "pending", + "blockedBy": [1, 2], + "description": "**Goal:** Map existing pure per-row compute over a ProcessPoolExecutor, assemble in input order, keep sort + CID-quantization boundary, sequential fallback for small inputs/workers=1 — output byte-identical regardless of worker count.\n\n**Files:** Modify software/src/pipeline.py (resolve_workers, _pmap, _peptide_worker, _antibody_worker; thread workers through run/run_peptide/run_antibody_tcr — lines 71-91, 159-221, 414-439); Test software/tests/unit/test_parallel_invariance.py\n\n**Verify:** uv run pytest tests/unit/test_parallel_invariance.py tests/regression/ -v\n\n```json:metadata\n{\"files\": [\"software/src/pipeline.py\",\"software/tests/unit/test_parallel_invariance.py\"], \"verifyCommand\": \"uv run pytest tests/unit/test_parallel_invariance.py tests/regression/ -v\", \"acceptanceCriteria\": [\"workers=1 == workers=4 byte-identical\",\"golden master unchanged\",\"worker fns picklable/spawn-safe\",\"results in input order\"]}\n```" + }, + { + "id": 4, + "subject": "Task 4: Resolve worker count from environment in CLI", + "status": "pending", + "blockedBy": [3], + "description": "**Goal:** CLI uses allocated cores at runtime without adding to argv — exec node command line stays constant so its CID does not change.\n\n**Files:** Modify software/src/main.py (import line 19; lines 44-46); Test software/tests/integration/test_cli.py\n\n**Verify:** uv run pytest tests/integration/test_cli.py -v\n\n```json:metadata\n{\"files\": [\"software/src/main.py\",\"software/tests/integration/test_cli.py\"], \"verifyCommand\": \"uv run pytest tests/integration/test_cli.py -v\", \"acceptanceCriteria\": [\"workers resolved from env, not argv\",\"argv unchanged (CID stable)\",\"PL_COMPUTE_WORKERS does not change output bytes\"]}\n```" + }, + { + "id": 5, + "subject": "Task 5: Request more cores + memory on workflow exec step", + "status": "pending", + "blockedBy": [4], + "description": "**Goal:** Schedule the Python step with multiple cores + more memory; keep exec node identity stable.\n\n**Files:** Modify workflow/src/main.tpl.tengo (pyRun, lines 344-360): mem 4GiB->8GiB, cpu(1)->cpu(8); Create .changeset/seqprops-parallel-compute.md (.software patch + .workflow patch); Verify against core/platforma/sdk/workflow-tengo/src/exec/index.lib.tengo\n\n**Verify:** pnpm run build:dev (block root)\n\n```json:metadata\n{\"files\": [\"workflow/src/main.tpl.tengo\",\".changeset/seqprops-parallel-compute.md\"], \"verifyCommand\": \"pnpm run build:dev\", \"acceptanceCriteria\": [\"pyRun .cpu(8) + .mem(8GiB)\",\"no machine-dependent argv added\",\"cpu/mem-vs-CID verified against SDK\",\"build:dev succeeds\",\"changeset for workflow+software\"]}\n```" + }, + { + "id": 6, + "subject": "Task 6 (OPTIONAL): Streaming compute path to bound memory", + "status": "pending", + "blockedBy": [5], + "description": "**OPTIONAL — decide at end.** Do not implement until Phase B merged + measured; the 8GiB bump may make this unnecessary.\n\n**Goal:** Process input in sorted batches, write output incrementally — peak memory O(batch). Output byte-identical (golden master is guard).\n\n**Files:** Modify software/src/io_layer.py, software/src/pipeline.py; Test software/tests/regression/test_streaming_equivalence.py\n\n**Verify (only if implemented):** uv run pytest tests/regression/test_streaming_equivalence.py tests/regression/test_golden_master.py -v\n\n```json:metadata\n{\"files\": [\"software/src/io_layer.py\",\"software/src/pipeline.py\",\"software/tests/regression/test_streaming_equivalence.py\"], \"verifyCommand\": \"uv run pytest tests/regression/test_streaming_equivalence.py tests/regression/test_golden_master.py -v\", \"acceptanceCriteria\": [\"decision gate recorded with RSS measurement\",\"streaming output byte-identical to in-memory\",\"golden master still passes\",\"env-gated, argv unchanged\"], \"optional\": true}\n```" + } + ], + "lastUpdated": "2026-06-10" +} From f73de89045afa28e10635b3951b06dac2a1507ab Mon Sep 17 00:00:00 2001 From: Paul Newling Date: Wed, 10 Jun 2026 09:08:34 -0700 Subject: [PATCH 02/10] test: freeze full-output golden master for sequence-properties refactor --- .../golden/antibody_cdr3_only/aa_fraction.tsv | 1 + .../data/golden/antibody_cdr3_only/input.tsv | 3 + .../data/golden/antibody_cdr3_only/plan.json | 1 + .../golden/antibody_cdr3_only/properties.tsv | 3 + .../data/golden/antibody_cdr3_only/stats.json | 1 + .../data/golden/antibody_full/aa_fraction.tsv | 1 + .../tests/data/golden/antibody_full/input.tsv | 4 + .../tests/data/golden/antibody_full/plan.json | 1 + .../data/golden/antibody_full/properties.tsv | 4 + .../data/golden/antibody_full/stats.json | 1 + .../golden/antibody_partial/aa_fraction.tsv | 1 + .../data/golden/antibody_partial/input.tsv | 2 + .../data/golden/antibody_partial/plan.json | 1 + .../golden/antibody_partial/properties.tsv | 2 + .../data/golden/antibody_partial/stats.json | 1 + .../tests/data/golden/peptide/aa_fraction.tsv | 201 ++++++++++++++++++ software/tests/data/golden/peptide/input.tsv | 11 + software/tests/data/golden/peptide/plan.json | 1 + .../tests/data/golden/peptide/properties.tsv | 11 + software/tests/data/golden/peptide/stats.json | 1 + .../tests/data/golden/tcr/aa_fraction.tsv | 1 + software/tests/data/golden/tcr/input.tsv | 2 + software/tests/data/golden/tcr/plan.json | 1 + software/tests/data/golden/tcr/properties.tsv | 2 + software/tests/data/golden/tcr/stats.json | 1 + software/tests/regression/__init__.py | 0 software/tests/regression/conftest.py | 11 + software/tests/regression/golden_cases.py | 100 +++++++++ software/tests/regression/regen_golden.py | 62 ++++++ .../tests/regression/test_golden_master.py | 74 +++++++ 30 files changed, 506 insertions(+) create mode 100644 software/tests/data/golden/antibody_cdr3_only/aa_fraction.tsv create mode 100644 software/tests/data/golden/antibody_cdr3_only/input.tsv create mode 100644 software/tests/data/golden/antibody_cdr3_only/plan.json create mode 100644 software/tests/data/golden/antibody_cdr3_only/properties.tsv create mode 100644 software/tests/data/golden/antibody_cdr3_only/stats.json create mode 100644 software/tests/data/golden/antibody_full/aa_fraction.tsv create mode 100644 software/tests/data/golden/antibody_full/input.tsv create mode 100644 software/tests/data/golden/antibody_full/plan.json create mode 100644 software/tests/data/golden/antibody_full/properties.tsv create mode 100644 software/tests/data/golden/antibody_full/stats.json create mode 100644 software/tests/data/golden/antibody_partial/aa_fraction.tsv create mode 100644 software/tests/data/golden/antibody_partial/input.tsv create mode 100644 software/tests/data/golden/antibody_partial/plan.json create mode 100644 software/tests/data/golden/antibody_partial/properties.tsv create mode 100644 software/tests/data/golden/antibody_partial/stats.json create mode 100644 software/tests/data/golden/peptide/aa_fraction.tsv create mode 100644 software/tests/data/golden/peptide/input.tsv create mode 100644 software/tests/data/golden/peptide/plan.json create mode 100644 software/tests/data/golden/peptide/properties.tsv create mode 100644 software/tests/data/golden/peptide/stats.json create mode 100644 software/tests/data/golden/tcr/aa_fraction.tsv create mode 100644 software/tests/data/golden/tcr/input.tsv create mode 100644 software/tests/data/golden/tcr/plan.json create mode 100644 software/tests/data/golden/tcr/properties.tsv create mode 100644 software/tests/data/golden/tcr/stats.json create mode 100644 software/tests/regression/__init__.py create mode 100644 software/tests/regression/conftest.py create mode 100644 software/tests/regression/golden_cases.py create mode 100644 software/tests/regression/regen_golden.py create mode 100644 software/tests/regression/test_golden_master.py diff --git a/software/tests/data/golden/antibody_cdr3_only/aa_fraction.tsv b/software/tests/data/golden/antibody_cdr3_only/aa_fraction.tsv new file mode 100644 index 0000000..fac5371 --- /dev/null +++ b/software/tests/data/golden/antibody_cdr3_only/aa_fraction.tsv @@ -0,0 +1 @@ +entity_key aminoAcid value diff --git a/software/tests/data/golden/antibody_cdr3_only/input.tsv b/software/tests/data/golden/antibody_cdr3_only/input.tsv new file mode 100644 index 0000000..72af48a --- /dev/null +++ b/software/tests/data/golden/antibody_cdr3_only/input.tsv @@ -0,0 +1,3 @@ +entity_key A_CDR3 B_CDR3 +c1 CARDYW CQQYNS +c2 CARGFW CQHFSS diff --git a/software/tests/data/golden/antibody_cdr3_only/plan.json b/software/tests/data/golden/antibody_cdr3_only/plan.json new file mode 100644 index 0000000..c426c63 --- /dev/null +++ b/software/tests/data/golden/antibody_cdr3_only/plan.json @@ -0,0 +1 @@ +{"mode": "antibody_tcr_legacy_sc", "receptor": "IG", "chains": ["A", "B"], "fullChains": [], "hasFv": false} \ No newline at end of file diff --git a/software/tests/data/golden/antibody_cdr3_only/properties.tsv b/software/tests/data/golden/antibody_cdr3_only/properties.tsv new file mode 100644 index 0000000..bdeb418 --- /dev/null +++ b/software/tests/data/golden/antibody_cdr3_only/properties.tsv @@ -0,0 +1,3 @@ +entity_key charge_A_CDR3 chargeShift_A_CDR3 gravy_A_CDR3 charge_B_CDR3 chargeShift_B_CDR3 gravy_B_CDR3 +c1 -0.111 -0.245 -0.9833333333333334 -0.112 -0.236 -1.6833333333333333 +c2 0.895 -0.22 0.21666666666666665 0.111 -0.854 -0.5000000000000001 diff --git a/software/tests/data/golden/antibody_cdr3_only/stats.json b/software/tests/data/golden/antibody_cdr3_only/stats.json new file mode 100644 index 0000000..b54a3d7 --- /dev/null +++ b/software/tests/data/golden/antibody_cdr3_only/stats.json @@ -0,0 +1 @@ +{"medianCdr3Length":{"A":6.0,"B":6.0}} \ No newline at end of file diff --git a/software/tests/data/golden/antibody_full/aa_fraction.tsv b/software/tests/data/golden/antibody_full/aa_fraction.tsv new file mode 100644 index 0000000..fac5371 --- /dev/null +++ b/software/tests/data/golden/antibody_full/aa_fraction.tsv @@ -0,0 +1 @@ +entity_key aminoAcid value diff --git a/software/tests/data/golden/antibody_full/input.tsv b/software/tests/data/golden/antibody_full/input.tsv new file mode 100644 index 0000000..080e6e6 --- /dev/null +++ b/software/tests/data/golden/antibody_full/input.tsv @@ -0,0 +1,4 @@ +entity_key A_FR1 A_CDR1 A_FR2 A_CDR2 A_FR3 A_CDR3 A_FR4 B_FR1 B_CDR1 B_FR2 B_CDR2 B_FR3 B_CDR3 B_FR4 +c1 EVQLVES GFTFSSY AMSWVRQ ISGSGGS TYYAESVKGRFTI CARDYW WGQGTLV DIQMTQS QSISSY LNWYQQK AASSLQS GVPSRFSGSG CQQYNS FGQGTKV +c2 EVQLVES GFTFSSY AMSWVRQ ISGSGGS CARGFW WGQGTLV DIQMTQS QSISSY LNWYQQK AASSLQS GVPSRFSGSG CQHFSS FGQGTKV +c3 EVQLVES GFTFSSY AMSWVRQ ISGSGGS TYYAESVKGRFTI WGQGTLV DIQMTQS QSISSY LNWYQQK AASSLQS GVPSRFSGSG CQQYNS FGQGTKV diff --git a/software/tests/data/golden/antibody_full/plan.json b/software/tests/data/golden/antibody_full/plan.json new file mode 100644 index 0000000..5203857 --- /dev/null +++ b/software/tests/data/golden/antibody_full/plan.json @@ -0,0 +1 @@ +{"mode": "antibody_tcr_legacy_bulk", "receptor": "IG", "chains": ["A", "B"], "fullChains": ["A", "B"], "hasFv": true} \ No newline at end of file diff --git a/software/tests/data/golden/antibody_full/properties.tsv b/software/tests/data/golden/antibody_full/properties.tsv new file mode 100644 index 0000000..482c3d3 --- /dev/null +++ b/software/tests/data/golden/antibody_full/properties.tsv @@ -0,0 +1,4 @@ +entity_key charge_A_CDR3 chargeShift_A_CDR3 gravy_A_CDR3 charge_B_CDR3 chargeShift_B_CDR3 gravy_B_CDR3 charge_A_VDJRegion pi_A_VDJRegion gravy_A_VDJRegion mw_A_VDJRegion eox_A_VDJRegion ered_A_VDJRegion instability_A_VDJRegion aliphatic_A_VDJRegion aromaticity_A_VDJRegion charge_B_VDJRegion pi_B_VDJRegion gravy_B_VDJRegion mw_B_VDJRegion eox_B_VDJRegion ered_B_VDJRegion instability_B_VDJRegion aliphatic_B_VDJRegion aromaticity_B_VDJRegion charge_Fv chargeShift_Fv pi_Fv eox_Fv ered_Fv mw_Fv +c1 -0.111 -0.245 -0.9833333333333334 -0.112 -0.236 -1.6833333333333333 -0.837 6.007 -0.1111111111111111 6050.6567 22460.0 22460.0 38.7537037037037 61.29629629629629 0.18518518518518517 1.149 9.166 -0.686 5466.8949999999995 9970.0 9970.0 62.929999999999986 46.8 0.12 0.313 -1.836 7.634 32430.0 32430.0 11517.5517 +c2 0.895 -0.22 0.21666666666666665 0.111 -0.854 -0.5000000000000001 1.18 9.169 -0.544 5432.8804 8480.0 8480.0 62.36399999999999 46.8 0.12 +c3 -0.112 -0.236 -1.6833333333333333 1.149 9.166 -0.686 5466.8949999999995 9970.0 9970.0 62.929999999999986 46.8 0.12 diff --git a/software/tests/data/golden/antibody_full/stats.json b/software/tests/data/golden/antibody_full/stats.json new file mode 100644 index 0000000..b54a3d7 --- /dev/null +++ b/software/tests/data/golden/antibody_full/stats.json @@ -0,0 +1 @@ +{"medianCdr3Length":{"A":6.0,"B":6.0}} \ No newline at end of file diff --git a/software/tests/data/golden/antibody_partial/aa_fraction.tsv b/software/tests/data/golden/antibody_partial/aa_fraction.tsv new file mode 100644 index 0000000..fac5371 --- /dev/null +++ b/software/tests/data/golden/antibody_partial/aa_fraction.tsv @@ -0,0 +1 @@ +entity_key aminoAcid value diff --git a/software/tests/data/golden/antibody_partial/input.tsv b/software/tests/data/golden/antibody_partial/input.tsv new file mode 100644 index 0000000..d394c70 --- /dev/null +++ b/software/tests/data/golden/antibody_partial/input.tsv @@ -0,0 +1,2 @@ +entity_key A_FR1 A_CDR1 A_FR2 A_CDR2 A_FR3 A_CDR3 A_FR4 B_FR1 B_CDR1 B_FR2 B_CDR2 B_FR3 B_CDR3 B_FR4 +c1 EVQLVES GFTFSSY AMSWVRQ ISGSGGS TYYAESVKGRFTI CARDYW WGQGTLV DIQMTQS QSISSY LNWYQQK AASSLQS GVPSRFSGSG CQQYNS FGQGTKV diff --git a/software/tests/data/golden/antibody_partial/plan.json b/software/tests/data/golden/antibody_partial/plan.json new file mode 100644 index 0000000..581df99 --- /dev/null +++ b/software/tests/data/golden/antibody_partial/plan.json @@ -0,0 +1 @@ +{"mode": "antibody_tcr_legacy_bulk", "receptor": "IG", "chains": ["A", "B"], "fullChains": ["A"], "hasFv": false} \ No newline at end of file diff --git a/software/tests/data/golden/antibody_partial/properties.tsv b/software/tests/data/golden/antibody_partial/properties.tsv new file mode 100644 index 0000000..ccaf887 --- /dev/null +++ b/software/tests/data/golden/antibody_partial/properties.tsv @@ -0,0 +1,2 @@ +entity_key charge_A_CDR3 chargeShift_A_CDR3 gravy_A_CDR3 charge_B_CDR3 chargeShift_B_CDR3 gravy_B_CDR3 charge_A_VDJRegion pi_A_VDJRegion gravy_A_VDJRegion mw_A_VDJRegion eox_A_VDJRegion ered_A_VDJRegion instability_A_VDJRegion aliphatic_A_VDJRegion aromaticity_A_VDJRegion +c1 -0.111 -0.245 -0.9833333333333334 -0.112 -0.236 -1.6833333333333333 -0.837 6.007 -0.1111111111111111 6050.6567 22460.0 22460.0 38.7537037037037 61.29629629629629 0.18518518518518517 diff --git a/software/tests/data/golden/antibody_partial/stats.json b/software/tests/data/golden/antibody_partial/stats.json new file mode 100644 index 0000000..b54a3d7 --- /dev/null +++ b/software/tests/data/golden/antibody_partial/stats.json @@ -0,0 +1 @@ +{"medianCdr3Length":{"A":6.0,"B":6.0}} \ No newline at end of file diff --git a/software/tests/data/golden/peptide/aa_fraction.tsv b/software/tests/data/golden/peptide/aa_fraction.tsv new file mode 100644 index 0000000..50ff950 --- /dev/null +++ b/software/tests/data/golden/peptide/aa_fraction.tsv @@ -0,0 +1,201 @@ +entity_key aminoAcid value +p_acidic A 0.0 +p_acidic C 0.0 +p_acidic D 0.4 +p_acidic E 0.6 +p_acidic F 0.0 +p_acidic G 0.0 +p_acidic H 0.0 +p_acidic I 0.0 +p_acidic K 0.0 +p_acidic L 0.0 +p_acidic M 0.0 +p_acidic N 0.0 +p_acidic P 0.0 +p_acidic Q 0.0 +p_acidic R 0.0 +p_acidic S 0.0 +p_acidic T 0.0 +p_acidic V 0.0 +p_acidic W 0.0 +p_acidic Y 0.0 +p_all_nonstd A +p_all_nonstd C +p_all_nonstd D +p_all_nonstd E +p_all_nonstd F +p_all_nonstd G +p_all_nonstd H +p_all_nonstd I +p_all_nonstd K +p_all_nonstd L +p_all_nonstd M +p_all_nonstd N +p_all_nonstd P +p_all_nonstd Q +p_all_nonstd R +p_all_nonstd S +p_all_nonstd T +p_all_nonstd V +p_all_nonstd W +p_all_nonstd Y +p_basic A 0.0 +p_basic C 0.0 +p_basic D 0.0 +p_basic E 0.0 +p_basic F 0.0 +p_basic G 0.0 +p_basic H 0.6 +p_basic I 0.0 +p_basic K 0.4 +p_basic L 0.0 +p_basic M 0.0 +p_basic N 0.0 +p_basic P 0.0 +p_basic Q 0.0 +p_basic R 0.0 +p_basic S 0.0 +p_basic T 0.0 +p_basic V 0.0 +p_basic W 0.0 +p_basic Y 0.0 +p_empty A +p_empty C +p_empty D +p_empty E +p_empty F +p_empty G +p_empty H +p_empty I +p_empty K +p_empty L +p_empty M +p_empty N +p_empty P +p_empty Q +p_empty R +p_empty S +p_empty T +p_empty V +p_empty W +p_empty Y +p_no_aromatic A 1.0 +p_no_aromatic C 0.0 +p_no_aromatic D 0.0 +p_no_aromatic E 0.0 +p_no_aromatic F 0.0 +p_no_aromatic G 0.0 +p_no_aromatic H 0.0 +p_no_aromatic I 0.0 +p_no_aromatic K 0.0 +p_no_aromatic L 0.0 +p_no_aromatic M 0.0 +p_no_aromatic N 0.0 +p_no_aromatic P 0.0 +p_no_aromatic Q 0.0 +p_no_aromatic R 0.0 +p_no_aromatic S 0.0 +p_no_aromatic T 0.0 +p_no_aromatic V 0.0 +p_no_aromatic W 0.0 +p_no_aromatic Y 0.0 +p_nonstd A 0.1111111111111111 +p_nonstd C 0.1111111111111111 +p_nonstd D 0.1111111111111111 +p_nonstd E 0.1111111111111111 +p_nonstd F 0.1111111111111111 +p_nonstd G 0.1111111111111111 +p_nonstd H 0.1111111111111111 +p_nonstd I 0.1111111111111111 +p_nonstd K 0.1111111111111111 +p_nonstd L 0.0 +p_nonstd M 0.0 +p_nonstd N 0.0 +p_nonstd P 0.0 +p_nonstd Q 0.0 +p_nonstd R 0.0 +p_nonstd S 0.0 +p_nonstd T 0.0 +p_nonstd V 0.0 +p_nonstd W 0.0 +p_nonstd Y 0.0 +p_paired_cys A 0.0 +p_paired_cys C 0.2222222222222222 +p_paired_cys D 0.0 +p_paired_cys E 0.0 +p_paired_cys F 0.0 +p_paired_cys G 0.1111111111111111 +p_paired_cys H 0.0 +p_paired_cys I 0.1111111111111111 +p_paired_cys K 0.0 +p_paired_cys L 0.1111111111111111 +p_paired_cys M 0.0 +p_paired_cys N 0.1111111111111111 +p_paired_cys P 0.1111111111111111 +p_paired_cys Q 0.1111111111111111 +p_paired_cys R 0.0 +p_paired_cys S 0.0 +p_paired_cys T 0.0 +p_paired_cys V 0.0 +p_paired_cys W 0.0 +p_paired_cys Y 0.1111111111111111 +p_short A 0.0 +p_short C 0.0 +p_short D 0.0 +p_short E 0.0 +p_short F 0.25 +p_short G 0.125 +p_short H 0.0 +p_short I 0.0 +p_short K 0.0 +p_short L 0.0 +p_short M 0.0 +p_short N 0.0 +p_short P 0.375 +p_short Q 0.0 +p_short R 0.125 +p_short S 0.125 +p_short T 0.0 +p_short V 0.0 +p_short W 0.0 +p_short Y 0.0 +p_stop A +p_stop C +p_stop D +p_stop E +p_stop F +p_stop G +p_stop H +p_stop I +p_stop K +p_stop L +p_stop M +p_stop N +p_stop P +p_stop Q +p_stop R +p_stop S +p_stop T +p_stop V +p_stop W +p_stop Y +p_valid A 0.1 +p_valid C 0.1 +p_valid D 0.1 +p_valid E 0.1 +p_valid F 0.1 +p_valid G 0.1 +p_valid H 0.1 +p_valid I 0.1 +p_valid K 0.1 +p_valid L 0.1 +p_valid M 0.0 +p_valid N 0.0 +p_valid P 0.0 +p_valid Q 0.0 +p_valid R 0.0 +p_valid S 0.0 +p_valid T 0.0 +p_valid V 0.0 +p_valid W 0.0 +p_valid Y 0.0 diff --git a/software/tests/data/golden/peptide/input.tsv b/software/tests/data/golden/peptide/input.tsv new file mode 100644 index 0000000..5ca5af5 --- /dev/null +++ b/software/tests/data/golden/peptide/input.tsv @@ -0,0 +1,11 @@ +entity_key sequence +p_valid ACDEFGHIKL +p_basic KKKKHHHHHH +p_acidic DDDDEEEEEE +p_short RPPGFSPF +p_no_aromatic AAAAAAAAAA +p_paired_cys CYIQNCPLG +p_nonstd ACDXEFGHIK +p_all_nonstd XXXXX +p_stop ACDE*GHIK +p_empty diff --git a/software/tests/data/golden/peptide/plan.json b/software/tests/data/golden/peptide/plan.json new file mode 100644 index 0000000..2ecb874 --- /dev/null +++ b/software/tests/data/golden/peptide/plan.json @@ -0,0 +1 @@ +{"mode": "peptide"} \ No newline at end of file diff --git a/software/tests/data/golden/peptide/properties.tsv b/software/tests/data/golden/peptide/properties.tsv new file mode 100644 index 0000000..fb49b9a --- /dev/null +++ b/software/tests/data/golden/peptide/properties.tsv @@ -0,0 +1,11 @@ +entity_key charge_peptide chargeShift_peptide gravy_peptide mw_peptide pi_peptide eox_peptide ered_peptide instability_peptide aliphatic_peptide aromaticity_peptide +p_acidic -10.078 -0.425 -3.5 1253.0489 2.953 0.0 0.0 172.0 0.0 0.0 +p_all_nonstd +p_basic 4.936 -4.577 -3.4799999999999995 1353.5403 8.745 0.0 0.0 9.0 0.0 0.0 +p_empty +p_no_aromatic -0.101 -0.211 1.8 728.7943 5.462 0.0 0.0 9.0 100.0 0.0 +p_nonstd -0.949 -1.033 -0.32222222222222224 1019.1317999999999 5.531 0.0 0.0 54.44444444444444 0.1111111111111111 +p_paired_cys -0.116 -0.245 0.33333333333333326 1010.1877999999999 5.436 1615.0 1490.0 86.66666666666666 0.1111111111111111 +p_short 0.899 -0.211 -0.6125 904.0228 9.72 0.0 0.0 0.0 0.25 +p_stop +p_valid -0.949 -1.033 0.08999999999999996 1132.2894 5.531 0.0 0.0 99.16000000000001 88.0 0.1 diff --git a/software/tests/data/golden/peptide/stats.json b/software/tests/data/golden/peptide/stats.json new file mode 100644 index 0000000..a3fd817 --- /dev/null +++ b/software/tests/data/golden/peptide/stats.json @@ -0,0 +1 @@ +{"hasPeptideBelowInstabilityFloor":true,"medianCdr3Length":{}} \ No newline at end of file diff --git a/software/tests/data/golden/tcr/aa_fraction.tsv b/software/tests/data/golden/tcr/aa_fraction.tsv new file mode 100644 index 0000000..fac5371 --- /dev/null +++ b/software/tests/data/golden/tcr/aa_fraction.tsv @@ -0,0 +1 @@ +entity_key aminoAcid value diff --git a/software/tests/data/golden/tcr/input.tsv b/software/tests/data/golden/tcr/input.tsv new file mode 100644 index 0000000..d394c70 --- /dev/null +++ b/software/tests/data/golden/tcr/input.tsv @@ -0,0 +1,2 @@ +entity_key A_FR1 A_CDR1 A_FR2 A_CDR2 A_FR3 A_CDR3 A_FR4 B_FR1 B_CDR1 B_FR2 B_CDR2 B_FR3 B_CDR3 B_FR4 +c1 EVQLVES GFTFSSY AMSWVRQ ISGSGGS TYYAESVKGRFTI CARDYW WGQGTLV DIQMTQS QSISSY LNWYQQK AASSLQS GVPSRFSGSG CQQYNS FGQGTKV diff --git a/software/tests/data/golden/tcr/plan.json b/software/tests/data/golden/tcr/plan.json new file mode 100644 index 0000000..beb3c72 --- /dev/null +++ b/software/tests/data/golden/tcr/plan.json @@ -0,0 +1 @@ +{"mode": "antibody_tcr_legacy_bulk", "receptor": "TCRAB", "chains": ["A", "B"], "fullChains": ["A", "B"], "hasFv": false} \ No newline at end of file diff --git a/software/tests/data/golden/tcr/properties.tsv b/software/tests/data/golden/tcr/properties.tsv new file mode 100644 index 0000000..67a847d --- /dev/null +++ b/software/tests/data/golden/tcr/properties.tsv @@ -0,0 +1,2 @@ +entity_key charge_A_CDR3 chargeShift_A_CDR3 gravy_A_CDR3 charge_B_CDR3 chargeShift_B_CDR3 gravy_B_CDR3 charge_A_VDJRegion pi_A_VDJRegion gravy_A_VDJRegion mw_A_VDJRegion eox_A_VDJRegion ered_A_VDJRegion instability_A_VDJRegion aliphatic_A_VDJRegion aromaticity_A_VDJRegion charge_B_VDJRegion pi_B_VDJRegion gravy_B_VDJRegion mw_B_VDJRegion eox_B_VDJRegion ered_B_VDJRegion instability_B_VDJRegion aliphatic_B_VDJRegion aromaticity_B_VDJRegion +c1 -0.111 -0.245 -0.9833333333333334 -0.112 -0.236 -1.6833333333333333 -0.837 6.007 -0.1111111111111111 6050.6567 22460.0 22460.0 38.7537037037037 61.29629629629629 0.18518518518518517 1.149 9.166 -0.686 5466.8949999999995 9970.0 9970.0 62.929999999999986 46.8 0.12 diff --git a/software/tests/data/golden/tcr/stats.json b/software/tests/data/golden/tcr/stats.json new file mode 100644 index 0000000..b54a3d7 --- /dev/null +++ b/software/tests/data/golden/tcr/stats.json @@ -0,0 +1 @@ +{"medianCdr3Length":{"A":6.0,"B":6.0}} \ No newline at end of file diff --git a/software/tests/regression/__init__.py b/software/tests/regression/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/software/tests/regression/conftest.py b/software/tests/regression/conftest.py new file mode 100644 index 0000000..e91a277 --- /dev/null +++ b/software/tests/regression/conftest.py @@ -0,0 +1,11 @@ +"""Put this regression package dir on sys.path so the byte-compare test can do +`from golden_cases import CASES` (a sibling module), matching regen_golden.py's +import. pytest auto-loads conftest before collecting sibling test modules. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) diff --git a/software/tests/regression/golden_cases.py b/software/tests/regression/golden_cases.py new file mode 100644 index 0000000..0022524 --- /dev/null +++ b/software/tests/regression/golden_cases.py @@ -0,0 +1,100 @@ +"""Golden-master case definitions — the single source of truth for both the +regeneration script and the byte-compare test. Each case is (input rows, +column order, plan dict). Inputs live in code; only the *outputs* are +committed under tests/data/golden//. + +Edge cases are drawn from the spec's documented behaviour tables (NA +propagation, non-standard residues, stop codon, per-clone missing region, +TCR has no Fv). +""" + +from __future__ import annotations + +_AB_COLS = ( + ["entity_key"] + + [f"A_{f}" for f in ("FR1", "CDR1", "FR2", "CDR2", "FR3", "CDR3", "FR4")] + + [f"B_{f}" for f in ("FR1", "CDR1", "FR2", "CDR2", "FR3", "CDR3", "FR4")] +) + +_AB_FULL_ROW = { + "entity_key": "c1", + "A_FR1": "EVQLVES", + "A_CDR1": "GFTFSSY", + "A_FR2": "AMSWVRQ", + "A_CDR2": "ISGSGGS", + "A_FR3": "TYYAESVKGRFTI", + "A_CDR3": "CARDYW", + "A_FR4": "WGQGTLV", + "B_FR1": "DIQMTQS", + "B_CDR1": "QSISSY", + "B_FR2": "LNWYQQK", + "B_CDR2": "AASSLQS", + "B_FR3": "GVPSRFSGSG", + "B_CDR3": "CQQYNS", + "B_FR4": "FGQGTKV", +} +# Clone with one heavy region missing -> full-chain A is NA for this clone only. +_AB_MISSING_REGION = {**_AB_FULL_ROW, "entity_key": "c2", "A_FR3": "", "A_CDR3": "CARGFW", "B_CDR3": "CQHFSS"} +# Clone with empty heavy CDR3 -> CDR3-A NA for this clone only. +_AB_EMPTY_CDR3 = {**_AB_FULL_ROW, "entity_key": "c3", "A_CDR3": ""} + +# CASES: name -> (rows, columns, plan) +CASES: dict[str, tuple[list[dict], list[str], dict]] = { + "peptide": ( + [ + {"entity_key": "p_valid", "sequence": "ACDEFGHIKL"}, + {"entity_key": "p_basic", "sequence": "KKKKHHHHHH"}, + {"entity_key": "p_acidic", "sequence": "DDDDEEEEEE"}, + {"entity_key": "p_short", "sequence": "RPPGFSPF"}, + {"entity_key": "p_no_aromatic", "sequence": "AAAAAAAAAA"}, + {"entity_key": "p_paired_cys", "sequence": "CYIQNCPLG"}, + {"entity_key": "p_nonstd", "sequence": "ACDXEFGHIK"}, + {"entity_key": "p_all_nonstd", "sequence": "XXXXX"}, + {"entity_key": "p_stop", "sequence": "ACDE*GHIK"}, + {"entity_key": "p_empty", "sequence": ""}, + ], + ["entity_key", "sequence"], + {"mode": "peptide"}, + ), + "antibody_full": ( + [_AB_FULL_ROW, _AB_MISSING_REGION, _AB_EMPTY_CDR3], + _AB_COLS, + { + "mode": "antibody_tcr_legacy_bulk", + "receptor": "IG", + "chains": ["A", "B"], + "fullChains": ["A", "B"], + "hasFv": True, + }, + ), + "antibody_cdr3_only": ( + [ + {"entity_key": "c1", "A_CDR3": "CARDYW", "B_CDR3": "CQQYNS"}, + {"entity_key": "c2", "A_CDR3": "CARGFW", "B_CDR3": "CQHFSS"}, + ], + ["entity_key", "A_CDR3", "B_CDR3"], + {"mode": "antibody_tcr_legacy_sc", "receptor": "IG", "chains": ["A", "B"], "fullChains": [], "hasFv": False}, + ), + "antibody_partial": ( + [_AB_FULL_ROW], + _AB_COLS, + { + "mode": "antibody_tcr_legacy_bulk", + "receptor": "IG", + "chains": ["A", "B"], + "fullChains": ["A"], + "hasFv": False, + }, + ), + "tcr": ( + [_AB_FULL_ROW], + _AB_COLS, + { + "mode": "antibody_tcr_legacy_bulk", + "receptor": "TCRAB", + "chains": ["A", "B"], + "fullChains": ["A", "B"], + "hasFv": False, + }, + ), +} diff --git a/software/tests/regression/regen_golden.py b/software/tests/regression/regen_golden.py new file mode 100644 index 0000000..96dcdba --- /dev/null +++ b/software/tests/regression/regen_golden.py @@ -0,0 +1,62 @@ +"""Regenerate committed golden outputs from the CURRENT code. + +Run ONLY deliberately: + uv run python tests/regression/regen_golden.py + +DO NOT run this during the parallelism/streaming refactor — if the golden +needs regenerating to make test_golden_master pass, the refactor changed +observable behaviour and that is a bug, not a snapshot update. Regenerate +only on an intentional behaviour change (BioPython bump, pKa update, spec +change) and call it out in the PR. +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "src")) + +from golden_cases import CASES # noqa: E402 + +from main import main # noqa: E402 + +GOLDEN_DIR = Path(__file__).resolve().parents[1] / "data" / "golden" + + +def _write_tsv(path: Path, rows: list[dict], columns: list[str]) -> None: + lines = ["\t".join(columns)] + for row in rows: + lines.append("\t".join(row.get(c, "") for c in columns)) + path.write_text("\n".join(lines) + "\n") + + +def regen() -> None: + for name, (rows, columns, plan) in CASES.items(): + case_dir = GOLDEN_DIR / name + case_dir.mkdir(parents=True, exist_ok=True) + in_tsv = case_dir / "input.tsv" + plan_json = case_dir / "plan.json" + _write_tsv(in_tsv, rows, columns) + plan_json.write_text(json.dumps(plan)) + rc = main( + [ + "--input", + str(in_tsv), + "--plan", + str(plan_json), + "--output", + str(case_dir / "properties.tsv"), + "--aa-fraction", + str(case_dir / "aa_fraction.tsv"), + "--stats", + str(case_dir / "stats.json"), + ] + ) + assert rc == 0, f"regen failed for case {name}" + print(f"regenerated golden/{name}") + + +if __name__ == "__main__": + regen() diff --git a/software/tests/regression/test_golden_master.py b/software/tests/regression/test_golden_master.py new file mode 100644 index 0000000..94b529e --- /dev/null +++ b/software/tests/regression/test_golden_master.py @@ -0,0 +1,74 @@ +"""Full-output byte snapshot — the refactor safety net. + +Each case runs main() into a tmp dir and asserts every output file is +byte-identical (sha256) to the committed golden produced by regen_golden.py +from pre-refactor code. If any later refactor changes a byte, this fails — +which is the entire point. +""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +import pytest +from golden_cases import CASES + +from main import main + +GOLDEN_DIR = Path(__file__).resolve().parents[1] / "data" / "golden" +_OUTPUTS = ("properties.tsv", "aa_fraction.tsv", "stats.json") + + +def _sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _write_tsv(path: Path, rows: list[dict], columns: list[str]) -> None: + lines = ["\t".join(columns)] + for row in rows: + lines.append("\t".join(row.get(c, "") for c in columns)) + path.write_text("\n".join(lines) + "\n") + + +@pytest.mark.parametrize("name", sorted(CASES), ids=sorted(CASES)) +def test_output_matches_golden(tmp_path: Path, name: str): + rows, columns, plan = CASES[name] + in_tsv = tmp_path / "input.tsv" + plan_json = tmp_path / "plan.json" + _write_tsv(in_tsv, rows, columns) + plan_json.write_text(json.dumps(plan)) + + out = {n: tmp_path / n for n in _OUTPUTS} + rc = main( + [ + "--input", + str(in_tsv), + "--plan", + str(plan_json), + "--output", + str(out["properties.tsv"]), + "--aa-fraction", + str(out["aa_fraction.tsv"]), + "--stats", + str(out["stats.json"]), + ] + ) + assert rc == 0 + + golden = GOLDEN_DIR / name + assert golden.is_dir(), f"missing golden dir for {name}; run regen_golden.py" + for n in _OUTPUTS: + assert _sha256(out[n]) == _sha256(golden / n), f"{name}/{n} diverged from golden — refactor changed behaviour" + + +# Non-vacuous guard: the antibody_full golden actually carries the Fv columns, +# and the tcr golden does NOT (R12). Proves the snapshot is meaningful, not empty. +def test_golden_column_presence(): + import polars as pl + + ab = pl.read_csv(GOLDEN_DIR / "antibody_full" / "properties.tsv", separator="\t") + assert {"charge_Fv", "pi_Fv", "charge_A_VDJRegion"} <= set(ab.columns) + tcr = pl.read_csv(GOLDEN_DIR / "tcr" / "properties.tsv", separator="\t") + assert not any(c.endswith("_Fv") for c in tcr.columns) From 4d973d1a7337b3661533e7a7859198fbfc852120 Mon Sep 17 00:00:00 2001 From: Paul Newling Date: Wed, 10 Jun 2026 09:15:53 -0700 Subject: [PATCH 03/10] test: pin row/key-preservation + byte-stability invariants at scale --- .../tests/regression/test_scale_invariants.py | 85 +++++++++++++++++++ 1 file changed, 85 insertions(+) create mode 100644 software/tests/regression/test_scale_invariants.py diff --git a/software/tests/regression/test_scale_invariants.py b/software/tests/regression/test_scale_invariants.py new file mode 100644 index 0000000..f6af5a3 --- /dev/null +++ b/software/tests/regression/test_scale_invariants.py @@ -0,0 +1,85 @@ +"""Structural invariants at scale — the per-row-map contract the parallel and +streaming refactors must preserve: one output row per input entity, same key +set, AA-fraction 20 rows/peptide, byte-stable across runs. + +Synthetic data is built from a SEEDED RNG (random.Random(0)) so the suite is +deterministic. Marked slow: 5000 rows is enough to exercise >1 worker chunk +once Phase B lands, but too slow for every iteration. +""" + +from __future__ import annotations + +import hashlib +import random +from pathlib import Path + +import polars as pl +import pytest + +from io_layer import write_output_tsv +from pipeline import run + +_AAS = "ACDEFGHIKLMNPQRSTVWY" +_REGIONS = ("FR1", "CDR1", "FR2", "CDR2", "FR3", "CDR3", "FR4") + + +def _rand_seq(rng: random.Random, lo: int, hi: int) -> str: + return "".join(rng.choice(_AAS) for _ in range(rng.randint(lo, hi))) + + +def _peptide_frame(n: int) -> pl.DataFrame: + rng = random.Random(0) + return pl.DataFrame( + {"entity_key": [f"p{i}" for i in range(n)], "sequence": [_rand_seq(rng, 5, 25) for _ in range(n)]}, + schema={"entity_key": pl.Utf8, "sequence": pl.Utf8}, + ) + + +def _antibody_frame(n: int) -> pl.DataFrame: + rng = random.Random(1) + cols: dict[str, list[str]] = {"entity_key": [f"c{i}" for i in range(n)]} + for ch in ("A", "B"): + for feat in _REGIONS: + cols[f"{ch}_{feat}"] = [_rand_seq(rng, 6, 14) for _ in range(n)] + schema = {k: pl.Utf8 for k in cols} + return pl.DataFrame(cols, schema=schema) + + +_AB_PLAN = { + "mode": "antibody_tcr_legacy_bulk", + "receptor": "IG", + "chains": ["A", "B"], + "fullChains": ["A", "B"], + "hasFv": True, +} + +N = 5000 + + +@pytest.mark.slow +def test_peptide_row_and_key_preservation(): + reads = _peptide_frame(N) + out = run(reads, {"mode": "peptide"}) + props = out["properties"] + assert props.height == N + assert set(props["entity_key"].to_list()) == set(reads["entity_key"].to_list()) + assert out["aa_fraction"].height == 20 * N + + +@pytest.mark.slow +def test_antibody_row_and_key_preservation(): + reads = _antibody_frame(N) + out = run(reads, _AB_PLAN) + props = out["properties"] + assert props.height == N + assert set(props["entity_key"].to_list()) == set(reads["entity_key"].to_list()) + assert "charge_Fv" in props.columns + + +@pytest.mark.slow +def test_peptide_byte_stable_two_runs(tmp_path: Path): + reads = _peptide_frame(N) + a, b = tmp_path / "a.tsv", tmp_path / "b.tsv" + write_output_tsv(run(reads, {"mode": "peptide"})["properties"], a, sort_keys=["entity_key"]) + write_output_tsv(run(reads, {"mode": "peptide"})["properties"], b, sort_keys=["entity_key"]) + assert hashlib.sha256(a.read_bytes()).hexdigest() == hashlib.sha256(b.read_bytes()).hexdigest() From 3082c24b23730b2770de0fa5674dbcdce1579115 Mon Sep 17 00:00:00 2001 From: Paul Newling Date: Wed, 10 Jun 2026 09:23:01 -0700 Subject: [PATCH 04/10] feat: parallelize per-row property compute across a process pool --- software/src/pipeline.py | 128 ++++++++++++------ .../tests/unit/test_parallel_invariance.py | 69 ++++++++++ 2 files changed, 155 insertions(+), 42 deletions(-) create mode 100644 software/tests/unit/test_parallel_invariance.py diff --git a/software/src/pipeline.py b/software/src/pipeline.py index 22742da..1932eaa 100644 --- a/software/src/pipeline.py +++ b/software/src/pipeline.py @@ -15,7 +15,10 @@ from __future__ import annotations +import functools import logging +import os +from concurrent.futures import ProcessPoolExecutor from typing import Any import polars as pl @@ -34,6 +37,39 @@ PH = 7.0 # All charge values computed at pH 7 (spec default). +# Parallelism. Below this row count the pool's process startup + pickle cost +# outweighs the benefit, so we stay in-process. The threshold also keeps the +# whole existing unit-test suite on the fast sequential path. +_PARALLEL_MIN_ROWS = 2000 + + +def resolve_workers(workers: int | None) -> int: + """How many worker processes to use. Explicit arg wins (used by tests and + by main.py once it reads the platform's CPU allocation). Falls back to the + PL_COMPUTE_WORKERS env var, then os.cpu_count(). The RESULT never depends on + this number — only the wall-clock does — so an over- or under-estimate is a + speed concern, never a correctness one. + """ + if workers is not None: + return max(1, int(workers)) + env = os.environ.get("PL_COMPUTE_WORKERS") + if env and env.isdigit() and int(env) > 0: + return int(env) + return max(1, os.cpu_count() or 1) + + +def _pmap(fn, items: list, workers: int, chunksize: int = 256) -> list: + """Map fn over items, preserving input order. Sequential below the + threshold or when workers<=1; otherwise a process pool. ProcessPoolExecutor + .map() preserves input order, so results align with items by index — the + property the byte-stable output depends on. + """ + if workers <= 1 or len(items) < _PARALLEL_MIN_ROWS: + return [fn(x) for x in items] + with ProcessPoolExecutor(max_workers=workers) as ex: + return list(ex.map(fn, items, chunksize=chunksize)) + + # --------------------------------------------------------------------------- # CID quantization # --------------------------------------------------------------------------- @@ -68,26 +104,23 @@ def _quantize_for_cid(df: pl.DataFrame) -> pl.DataFrame: # --------------------------------------------------------------------------- -def run(reads: pl.DataFrame, plan: dict[str, Any]) -> dict[str, Any]: - """Dispatch by mode. Returns a dict with three entries: - - - `properties` (DataFrame): one row per entity, columns per the plan. - - `aa_fraction` (DataFrame): long-format (entity_key, aminoAcid, value). - Empty body when mode is not peptide. - - `stats` (dict): dataset-level stats consumed by the workflow info layer - (e.g. R11c VHH detection — median CDR-H3 length per chain; - R9 — peptide count below the Instability Index length floor). +def run(reads: pl.DataFrame, plan: dict[str, Any], workers: int | None = None) -> dict[str, Any]: + """Dispatch by mode. `workers` controls parallelism only — output is + identical for any value (see test_parallel_invariance). Returns a dict with + `properties`, `aa_fraction`, and `stats` entries (unchanged contract). """ + n_workers = resolve_workers(workers) mode = plan["mode"] if mode == "peptide": - log.info("Running peptide mode (%d entities)", reads.height) - return run_peptide(reads) + log.info("Running peptide mode (%d entities, %d workers)", reads.height, n_workers) + return run_peptide(reads, n_workers) log.info( - "Running antibody/TCR mode (receptor=%s, %d clones)", + "Running antibody/TCR mode (receptor=%s, %d clones, %d workers)", plan.get("receptor", "IG"), reads.height, + n_workers, ) - return run_antibody_tcr(reads, plan) + return run_antibody_tcr(reads, plan, n_workers) # --------------------------------------------------------------------------- @@ -156,45 +189,48 @@ def _compute_peptide_row_from_ctx(ctx: SequenceContext) -> dict[str, float | Non } -def run_peptide(reads: pl.DataFrame) -> dict[str, Any]: - """Compute peptide-mode outputs. +def _peptide_worker(seq: str) -> tuple[dict[str, float | None], list[float | None] | None]: + """Picklable per-peptide unit: (properties row, 20 AA fractions in + STANDARD_AAS order | None). One SequenceContext per sequence, shared across + all 11 reads — same sharing the old inline loop relied on. + """ + ctx = SequenceContext.from_seq(seq) + if ctx is None: + return (dict(_NA_PEPTIDE_ROW), None) + props = _compute_peptide_row_from_ctx(ctx) + fr = ctx.aa_fractions() + return (props, [fr[aa] for aa in STANDARD_AAS]) + - Builds one `SequenceContext` per sequence and reuses it for both the - properties row and the AA-fraction rows — so each sequence is `_prepare`d - once and BioPython objects are constructed once across all 11 reads. - Accumulates columnar arrays directly into a dict-of-lists (one allocation - per column, vs. one dict per row) and constructs the DataFrame from those. +def run_peptide(reads: pl.DataFrame, workers: int = 1) -> dict[str, Any]: + """Compute peptide-mode outputs. Per-sequence work runs through `_pmap` + (sequential or pooled); results are reassembled in input order so the + serialized output is byte-identical regardless of worker count. """ keys = reads["entity_key"].to_list() seqs = reads["sequence"].to_list() n = len(seqs) log.info("Computing peptide properties + AA fractions (%d sequences)", n) - prop_cols: dict[str, list[Any]] = {"entity_key": [], **{c: [] for c in PEPTIDE_PROPERTY_COLUMNS}} + results = _pmap(_peptide_worker, seqs, workers) + + prop_cols: dict[str, list[Any]] = {"entity_key": list(keys), **{c: [] for c in PEPTIDE_PROPERTY_COLUMNS}} aa_entity: list[str] = [] aa_amino: list[str] = [] aa_value: list[float | None] = [] - for k, s in zip(keys, seqs): - prop_cols["entity_key"].append(k) - ctx = SequenceContext.from_seq(s) - if ctx is None: - for c in PEPTIDE_PROPERTY_COLUMNS: - prop_cols[c].append(None) - # Emit one row per std AA with NA value, so the 2-axis PColumn - # keeps a uniform shape across entities. + for k, (props, fractions) in zip(keys, results): + for c in PEPTIDE_PROPERTY_COLUMNS: + prop_cols[c].append(props[c]) + if fractions is None: for aa in STANDARD_AAS: aa_entity.append(k) aa_amino.append(aa) aa_value.append(None) else: - row = _compute_peptide_row_from_ctx(ctx) - for c in PEPTIDE_PROPERTY_COLUMNS: - prop_cols[c].append(row[c]) - fractions = ctx.aa_fractions() - for aa in STANDARD_AAS: + for aa, val in zip(STANDARD_AAS, fractions): aa_entity.append(k) aa_amino.append(aa) - aa_value.append(fractions[aa]) + aa_value.append(val) properties = pl.DataFrame( prop_cols, schema={"entity_key": pl.Utf8, **{c: pl.Float64 for c in PEPTIDE_PROPERTY_COLUMNS}}, @@ -204,10 +240,6 @@ def run_peptide(reads: pl.DataFrame) -> dict[str, Any]: schema={"entity_key": pl.Utf8, "aminoAcid": pl.Utf8, "value": pl.Float64}, ) - # R9 — flag whether any *real* peptide falls below the Instability Index - # floor. `if s` filters None / "" so the banner does not fire on empty - # cells (which are no peptide, not a short peptide); `0 < effective_length` - # filters sequences that clean to empty (e.g. all-non-standard residues). has_below_floor = any(0 < effective_length(s) < INSTABILITY_MIN_LENGTH for s in seqs if s) stats = { "medianCdr3Length": {}, @@ -411,7 +443,15 @@ def _median_cdr3_length_by_chain(reads: pl.DataFrame, chains: list[str]) -> dict return out -def run_antibody_tcr(reads: pl.DataFrame, plan: dict[str, Any]) -> dict[str, Any]: +def _antibody_worker(record: dict, plan: dict[str, Any]) -> dict[str, Any]: + """Picklable per-clone unit. `plan` is bound per-call via functools.partial; + it is a plain dict and pickles cleanly. Delegates to the existing + _compute_row_for so the computation is unchanged. + """ + return _compute_row_for(record, plan) + + +def run_antibody_tcr(reads: pl.DataFrame, plan: dict[str, Any], workers: int = 1) -> dict[str, Any]: chains = plan.get("chains", []) full_chains = plan.get("fullChains", []) n = reads.height @@ -421,10 +461,14 @@ def run_antibody_tcr(reads: pl.DataFrame, plan: dict[str, Any]) -> dict[str, Any log.info("Reconstructing full chains %s and computing full-chain properties", list(full_chains)) if plan.get("hasFv"): log.info("Computing Fv properties (paired VH+VL)") + out_cols = _planned_output_columns(plan) + records = list(reads.iter_rows(named=True)) + worker = functools.partial(_antibody_worker, plan=plan) + rows = _pmap(worker, records, workers) + columns: dict[str, list[Any]] = {"entity_key": [], **{c: [] for c in out_cols}} - for record in reads.iter_rows(named=True): - row = _compute_row_for(record, plan) + for row in rows: columns["entity_key"].append(row["entity_key"]) for c in out_cols: columns[c].append(row[c]) diff --git a/software/tests/unit/test_parallel_invariance.py b/software/tests/unit/test_parallel_invariance.py new file mode 100644 index 0000000..7b2c69d --- /dev/null +++ b/software/tests/unit/test_parallel_invariance.py @@ -0,0 +1,69 @@ +"""Worker-count invariance — the core safety property for parallelism. + +The output of run() must not depend on how many workers compute it. We +compare workers=1 against workers=4 (and a larger frame that actually spills +into the pool) and assert byte-identical serialized output + identical stats. +""" + +from __future__ import annotations + +import random +from pathlib import Path + +import polars as pl +import pytest + +from io_layer import write_output_tsv +from pipeline import run + +_AAS = "ACDEFGHIKLMNPQRSTVWY" +_REGIONS = ("FR1", "CDR1", "FR2", "CDR2", "FR3", "CDR3", "FR4") +_AB_PLAN = { + "mode": "antibody_tcr_legacy_bulk", + "receptor": "IG", + "chains": ["A", "B"], + "fullChains": ["A", "B"], + "hasFv": True, +} + + +def _peptide_frame(n: int, seed: int) -> pl.DataFrame: + rng = random.Random(seed) + seqs = ["".join(rng.choice(_AAS) for _ in range(rng.randint(5, 25))) for _ in range(n)] + return pl.DataFrame( + {"entity_key": [f"p{i}" for i in range(n)], "sequence": seqs}, + schema={"entity_key": pl.Utf8, "sequence": pl.Utf8}, + ) + + +def _antibody_frame(n: int, seed: int) -> pl.DataFrame: + rng = random.Random(seed) + cols = {"entity_key": [f"c{i}" for i in range(n)]} + for ch in ("A", "B"): + for feat in _REGIONS: + cols[f"{ch}_{feat}"] = ["".join(rng.choice(_AAS) for _ in range(rng.randint(6, 14))) for _ in range(n)] + return pl.DataFrame(cols, schema={k: pl.Utf8 for k in cols}) + + +def _serialize(out: dict, tmp: Path, tag: str) -> tuple[bytes, bytes, dict]: + p = tmp / f"props_{tag}.tsv" + a = tmp / f"aa_{tag}.tsv" + write_output_tsv(out["properties"], p, sort_keys=["entity_key"]) + write_output_tsv(out["aa_fraction"], a, sort_keys=["entity_key", "aminoAcid"]) + return p.read_bytes(), a.read_bytes(), out["stats"] + + +@pytest.mark.parametrize("n", [3, 3000], ids=["sequential-path", "pool-path"]) +def test_peptide_invariant_to_workers(tmp_path: Path, n: int): + reads = _peptide_frame(n, seed=0) + one = _serialize(run(reads, {"mode": "peptide"}, workers=1), tmp_path, "1") + four = _serialize(run(reads, {"mode": "peptide"}, workers=4), tmp_path, "4") + assert one == four + + +@pytest.mark.parametrize("n", [3, 3000], ids=["sequential-path", "pool-path"]) +def test_antibody_invariant_to_workers(tmp_path: Path, n: int): + reads = _antibody_frame(n, seed=1) + one = _serialize(run(reads, _AB_PLAN, workers=1), tmp_path, "1") + four = _serialize(run(reads, _AB_PLAN, workers=4), tmp_path, "4") + assert one == four From edbe38af366917e4365a680e43986972a8a0b00c Mon Sep 17 00:00:00 2001 From: Paul Newling Date: Wed, 10 Jun 2026 09:29:46 -0700 Subject: [PATCH 05/10] refactor: surface compute-pool worker failures with context; test exception propagation --- software/src/pipeline.py | 20 +++++++++++++++++-- .../tests/unit/test_parallel_invariance.py | 13 ++++++++++++ 2 files changed, 31 insertions(+), 2 deletions(-) diff --git a/software/src/pipeline.py b/software/src/pipeline.py index 1932eaa..c155266 100644 --- a/software/src/pipeline.py +++ b/software/src/pipeline.py @@ -19,6 +19,7 @@ import logging import os from concurrent.futures import ProcessPoolExecutor +from concurrent.futures.process import BrokenProcessPool from typing import Any import polars as pl @@ -40,8 +41,16 @@ # Parallelism. Below this row count the pool's process startup + pickle cost # outweighs the benefit, so we stay in-process. The threshold also keeps the # whole existing unit-test suite on the fast sequential path. +# The threshold is compared against ROW count regardless of per-row cost — an +# antibody/TCR clone does several times the work of a peptide (CDR3 × chains + +# full-chain × chains + Fv), so 2000 is intentionally more conservative for +# antibody mode than peptide mode. _PARALLEL_MIN_ROWS = 2000 +# Rows dispatched per pool task. Bounds per-task pickle/IPC overhead without +# starving workers; never overridden today but kept as a tuning knob. +_PARALLEL_CHUNKSIZE = 256 + def resolve_workers(workers: int | None) -> int: """How many worker processes to use. Explicit arg wins (used by tests and @@ -58,7 +67,7 @@ def resolve_workers(workers: int | None) -> int: return max(1, os.cpu_count() or 1) -def _pmap(fn, items: list, workers: int, chunksize: int = 256) -> list: +def _pmap(fn, items: list, workers: int, chunksize: int = _PARALLEL_CHUNKSIZE) -> list: """Map fn over items, preserving input order. Sequential below the threshold or when workers<=1; otherwise a process pool. ProcessPoolExecutor .map() preserves input order, so results align with items by index — the @@ -67,7 +76,14 @@ def _pmap(fn, items: list, workers: int, chunksize: int = 256) -> list: if workers <= 1 or len(items) < _PARALLEL_MIN_ROWS: return [fn(x) for x in items] with ProcessPoolExecutor(max_workers=workers) as ex: - return list(ex.map(fn, items, chunksize=chunksize)) + try: + return list(ex.map(fn, items, chunksize=chunksize)) + except BrokenProcessPool as exc: + raise RuntimeError( + f"compute worker pool broke while processing {len(items)} items " + f"with {workers} workers — a worker process was killed (most likely " + f"out of memory). Reduce the input size or raise the step's memory." + ) from exc # --------------------------------------------------------------------------- diff --git a/software/tests/unit/test_parallel_invariance.py b/software/tests/unit/test_parallel_invariance.py index 7b2c69d..174ae82 100644 --- a/software/tests/unit/test_parallel_invariance.py +++ b/software/tests/unit/test_parallel_invariance.py @@ -67,3 +67,16 @@ def test_antibody_invariant_to_workers(tmp_path: Path, n: int): one = _serialize(run(reads, _AB_PLAN, workers=1), tmp_path, "1") four = _serialize(run(reads, _AB_PLAN, workers=4), tmp_path, "4") assert one == four + + +def _raise_on_call(_x): + raise ValueError("worker boom") + + +def test_pmap_surfaces_clean_worker_exception(): + # A worker that raises must propagate the original exception to the caller, + # not hang or be swallowed. >= _PARALLEL_MIN_ROWS forces the pool path. + from pipeline import _pmap + + with pytest.raises(ValueError): + _pmap(_raise_on_call, list(range(2500)), workers=4) From 8c0aaa3b15a7d29ecafcad4969e162b488900f75 Mon Sep 17 00:00:00 2001 From: Paul Newling Date: Wed, 10 Jun 2026 09:32:56 -0700 Subject: [PATCH 06/10] feat: resolve compute worker count from environment in CLI --- software/src/main.py | 6 ++++-- software/tests/integration/test_cli.py | 22 ++++++++++++++++++++++ 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/software/src/main.py b/software/src/main.py index 422f7bf..f6ce514 100644 --- a/software/src/main.py +++ b/software/src/main.py @@ -16,7 +16,7 @@ from pathlib import Path from io_layer import read_input_tsv, read_plan, write_output_tsv -from pipeline import run +from pipeline import resolve_workers, run def _configure_logging() -> None: @@ -43,7 +43,9 @@ def main(argv: list[str] | None = None) -> int: reads = read_input_tsv(args.input) plan = read_plan(args.plan) - outputs = run(reads, plan) + workers = resolve_workers(None) + logging.getLogger(__name__).info("Using %d compute worker(s)", workers) + outputs = run(reads, plan, workers=workers) write_output_tsv(outputs["properties"], args.output, sort_keys=["entity_key"]) write_output_tsv(outputs["aa_fraction"], args.aa_fraction, sort_keys=["entity_key", "aminoAcid"]) Path(args.stats).write_text(json.dumps(outputs["stats"], sort_keys=True, separators=(",", ":"))) diff --git a/software/tests/integration/test_cli.py b/software/tests/integration/test_cli.py index 1114d6d..53766d2 100644 --- a/software/tests/integration/test_cli.py +++ b/software/tests/integration/test_cli.py @@ -10,6 +10,7 @@ from pathlib import Path import polars as pl +import pytest from main import main @@ -387,3 +388,24 @@ def test_cli_writes_progress_to_stderr(tmp_path: Path, capsys): captured = capsys.readouterr() assert "peptide" in captured.err.lower(), captured.err assert "scalar" in captured.err.lower() or "properties" in captured.err.lower(), captured.err + + +# Worker count comes from PL_COMPUTE_WORKERS (env), never argv, and must not +# change output bytes — only wall-clock. 2500 rows exceed the parallel +# threshold, so PL_COMPUTE_WORKERS=1 runs sequentially while =4 runs the pool: +# this proves the env knob selects the path AND the CLI output is invariant. +@pytest.mark.slow +def test_env_worker_count_does_not_change_output(tmp_path: Path, monkeypatch): + import random + + rng = random.Random(0) + aas = "ACDEFGHIKLMNPQRSTVWY" + rows = [ + {"entity_key": f"p{i}", "sequence": "".join(rng.choice(aas) for _ in range(rng.randint(5, 25)))} + for i in range(2500) + ] + monkeypatch.setenv("PL_COMPUTE_WORKERS", "1") + out_a, _ = _run_peptide(tmp_path, "_w1", rows) + monkeypatch.setenv("PL_COMPUTE_WORKERS", "4") + out_b, _ = _run_peptide(tmp_path, "_w4", rows) + assert _sha256(out_a) == _sha256(out_b) From 291eced5dbb787371bad40b09a1f533e9bfed1d0 Mon Sep 17 00:00:00 2001 From: Paul Newling Date: Wed, 10 Jun 2026 09:38:50 -0700 Subject: [PATCH 07/10] feat: request 8 cores + 8GiB for the sequence-properties compute step --- .changeset/seqprops-parallel-compute.md | 6 ++++++ workflow/src/main.tpl.tengo | 4 ++-- 2 files changed, 8 insertions(+), 2 deletions(-) create mode 100644 .changeset/seqprops-parallel-compute.md diff --git a/.changeset/seqprops-parallel-compute.md b/.changeset/seqprops-parallel-compute.md new file mode 100644 index 0000000..3ecd53a --- /dev/null +++ b/.changeset/seqprops-parallel-compute.md @@ -0,0 +1,6 @@ +--- +'@platforma-open/milaboratories.sequence-properties.software': patch +'@platforma-open/milaboratories.sequence-properties.workflow': patch +--- + +Parallelize the property computation across CPU cores and raise the compute step's CPU/memory request, so large datasets compute on the first run without the previous single-core bottleneck. Output is byte-identical to the prior version (worker count never affects results). diff --git a/workflow/src/main.tpl.tengo b/workflow/src/main.tpl.tengo index ef5afae..76ade41 100644 --- a/workflow/src/main.tpl.tengo +++ b/workflow/src/main.tpl.tengo @@ -344,8 +344,8 @@ wf.body(func(args) { soft := assets.importSoftware("@platforma-open/milaboratories.sequence-properties.software:compute-properties") pyRun := exec.builder(). software(soft). - mem("4GiB"). - cpu(1). + mem("8GiB"). + cpu(8). addFile("input.tsv", seqTable). writeFile("plan.json", canonical.encode(plan)). arg("--input").arg("input.tsv"). From b94a0e4157ed8054622da09b5495cd1759e6ebc4 Mon Sep 17 00:00:00 2001 From: Paul Newling Date: Wed, 10 Jun 2026 09:44:16 -0700 Subject: [PATCH 08/10] chore: bump changeset to minor (+root) and document compute step resource request --- .changeset/seqprops-parallel-compute.md | 5 +++-- workflow/src/main.tpl.tengo | 3 +++ 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/.changeset/seqprops-parallel-compute.md b/.changeset/seqprops-parallel-compute.md index 3ecd53a..6a90863 100644 --- a/.changeset/seqprops-parallel-compute.md +++ b/.changeset/seqprops-parallel-compute.md @@ -1,6 +1,7 @@ --- -'@platforma-open/milaboratories.sequence-properties.software': patch -'@platforma-open/milaboratories.sequence-properties.workflow': patch +'@platforma-open/milaboratories.sequence-properties.software': minor +'@platforma-open/milaboratories.sequence-properties.workflow': minor +'@platforma-open/milaboratories.sequence-properties': minor --- Parallelize the property computation across CPU cores and raise the compute step's CPU/memory request, so large datasets compute on the first run without the previous single-core bottleneck. Output is byte-identical to the prior version (worker count never affects results). diff --git a/workflow/src/main.tpl.tengo b/workflow/src/main.tpl.tengo index 76ade41..1bd2a0a 100644 --- a/workflow/src/main.tpl.tengo +++ b/workflow/src/main.tpl.tengo @@ -342,6 +342,9 @@ wf.body(func(args) { // stats.json (dataset-level scalars consumed by the info layer — e.g. R11c // median CDR-H3 length per chain). soft := assets.importSoftware("@platforma-open/milaboratories.sequence-properties.software:compute-properties") + // The Python step parallelizes per-row compute across os.cpu_count() workers + // (see software pipeline._pmap), so it requests multiple cores; ~1GiB/worker + // keeps the process pool from OOMing on large datasets. Other steps stay at cpu(1). pyRun := exec.builder(). software(soft). mem("8GiB"). From 2f614bc44e9c643c640c7d8a9190f30c0ca80372 Mon Sep 17 00:00:00 2001 From: Paul Newling Date: Wed, 10 Jun 2026 10:01:03 -0700 Subject: [PATCH 09/10] chore: log compute worker count once (drop redundant per-run fragment) --- software/src/pipeline.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/software/src/pipeline.py b/software/src/pipeline.py index c155266..d3a2254 100644 --- a/software/src/pipeline.py +++ b/software/src/pipeline.py @@ -128,13 +128,12 @@ def run(reads: pl.DataFrame, plan: dict[str, Any], workers: int | None = None) - n_workers = resolve_workers(workers) mode = plan["mode"] if mode == "peptide": - log.info("Running peptide mode (%d entities, %d workers)", reads.height, n_workers) + log.info("Running peptide mode (%d entities)", reads.height) return run_peptide(reads, n_workers) log.info( - "Running antibody/TCR mode (receptor=%s, %d clones, %d workers)", + "Running antibody/TCR mode (receptor=%s, %d clones)", plan.get("receptor", "IG"), reads.height, - n_workers, ) return run_antibody_tcr(reads, plan, n_workers) From b92d8adc8cab24f9724be1a499beed4222ba5c39 Mon Sep 17 00:00:00 2001 From: Paul Newling Date: Wed, 10 Jun 2026 10:02:22 -0700 Subject: [PATCH 10/10] docs: mark plan tasks 1-5 complete (task 6 streaming deferred) --- ...equence-properties-parallel-compute.md.tasks.json | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/docs/superpowers/plans/2026-06-10-sequence-properties-parallel-compute.md.tasks.json b/docs/superpowers/plans/2026-06-10-sequence-properties-parallel-compute.md.tasks.json index 8bdf518..44130f6 100644 --- a/docs/superpowers/plans/2026-06-10-sequence-properties-parallel-compute.md.tasks.json +++ b/docs/superpowers/plans/2026-06-10-sequence-properties-parallel-compute.md.tasks.json @@ -4,33 +4,33 @@ { "id": 1, "subject": "Task 1: Freeze full-output golden master", - "status": "pending", + "status": "completed", "description": "**Goal:** Committed byte-exact snapshot of full CLI output for every mode + coverage tier + edge case, generated from current pre-refactor code, asserted via sha256. Safety net for Phases B/C.\n\n**Files:** software/tests/regression/{__init__.py, golden_cases.py, regen_golden.py, test_golden_master.py}; software/tests/data/golden//{properties.tsv,aa_fraction.tsv,stats.json}\n\n**Verify:** uv run pytest tests/regression/test_golden_master.py -v\n\n```json:metadata\n{\"files\": [\"software/tests/regression/__init__.py\",\"software/tests/regression/golden_cases.py\",\"software/tests/regression/regen_golden.py\",\"software/tests/regression/test_golden_master.py\",\"software/tests/data/golden/\"], \"verifyCommand\": \"uv run pytest tests/regression/test_golden_master.py -v\", \"acceptanceCriteria\": [\"all golden cases byte-match via sha256\",\"cases cover all modes/tiers/edge-cases\",\"goldens generated from pre-refactor code and committed\"]}\n```" }, { "id": 2, "subject": "Task 2: Structural invariants at scale", - "status": "pending", + "status": "completed", "description": "**Goal:** Pin invariants the parallel/streaming refactor must preserve — one output row per input entity, same key set, AA-fraction 20 rows/peptide, two runs byte-identical — at 5000 rows.\n\n**Files:** software/tests/regression/test_scale_invariants.py\n\n**Verify:** uv run pytest tests/regression/test_scale_invariants.py -v\n\n```json:metadata\n{\"files\": [\"software/tests/regression/test_scale_invariants.py\"], \"verifyCommand\": \"uv run pytest tests/regression/test_scale_invariants.py -v\", \"acceptanceCriteria\": [\"row count + key set preserved\",\"aa_fraction 20 rows/peptide\",\"byte-stable across two runs\",\"seeded RNG, marked slow\"]}\n```" }, { "id": 3, "subject": "Task 3: Parallelize per-row compute across a process pool", - "status": "pending", + "status": "completed", "blockedBy": [1, 2], "description": "**Goal:** Map existing pure per-row compute over a ProcessPoolExecutor, assemble in input order, keep sort + CID-quantization boundary, sequential fallback for small inputs/workers=1 — output byte-identical regardless of worker count.\n\n**Files:** Modify software/src/pipeline.py (resolve_workers, _pmap, _peptide_worker, _antibody_worker; thread workers through run/run_peptide/run_antibody_tcr — lines 71-91, 159-221, 414-439); Test software/tests/unit/test_parallel_invariance.py\n\n**Verify:** uv run pytest tests/unit/test_parallel_invariance.py tests/regression/ -v\n\n```json:metadata\n{\"files\": [\"software/src/pipeline.py\",\"software/tests/unit/test_parallel_invariance.py\"], \"verifyCommand\": \"uv run pytest tests/unit/test_parallel_invariance.py tests/regression/ -v\", \"acceptanceCriteria\": [\"workers=1 == workers=4 byte-identical\",\"golden master unchanged\",\"worker fns picklable/spawn-safe\",\"results in input order\"]}\n```" }, { "id": 4, "subject": "Task 4: Resolve worker count from environment in CLI", - "status": "pending", + "status": "completed", "blockedBy": [3], "description": "**Goal:** CLI uses allocated cores at runtime without adding to argv — exec node command line stays constant so its CID does not change.\n\n**Files:** Modify software/src/main.py (import line 19; lines 44-46); Test software/tests/integration/test_cli.py\n\n**Verify:** uv run pytest tests/integration/test_cli.py -v\n\n```json:metadata\n{\"files\": [\"software/src/main.py\",\"software/tests/integration/test_cli.py\"], \"verifyCommand\": \"uv run pytest tests/integration/test_cli.py -v\", \"acceptanceCriteria\": [\"workers resolved from env, not argv\",\"argv unchanged (CID stable)\",\"PL_COMPUTE_WORKERS does not change output bytes\"]}\n```" }, { "id": 5, "subject": "Task 5: Request more cores + memory on workflow exec step", - "status": "pending", + "status": "completed", "blockedBy": [4], "description": "**Goal:** Schedule the Python step with multiple cores + more memory; keep exec node identity stable.\n\n**Files:** Modify workflow/src/main.tpl.tengo (pyRun, lines 344-360): mem 4GiB->8GiB, cpu(1)->cpu(8); Create .changeset/seqprops-parallel-compute.md (.software patch + .workflow patch); Verify against core/platforma/sdk/workflow-tengo/src/exec/index.lib.tengo\n\n**Verify:** pnpm run build:dev (block root)\n\n```json:metadata\n{\"files\": [\"workflow/src/main.tpl.tengo\",\".changeset/seqprops-parallel-compute.md\"], \"verifyCommand\": \"pnpm run build:dev\", \"acceptanceCriteria\": [\"pyRun .cpu(8) + .mem(8GiB)\",\"no machine-dependent argv added\",\"cpu/mem-vs-CID verified against SDK\",\"build:dev succeeds\",\"changeset for workflow+software\"]}\n```" }, @@ -42,5 +42,5 @@ "description": "**OPTIONAL — decide at end.** Do not implement until Phase B merged + measured; the 8GiB bump may make this unnecessary.\n\n**Goal:** Process input in sorted batches, write output incrementally — peak memory O(batch). Output byte-identical (golden master is guard).\n\n**Files:** Modify software/src/io_layer.py, software/src/pipeline.py; Test software/tests/regression/test_streaming_equivalence.py\n\n**Verify (only if implemented):** uv run pytest tests/regression/test_streaming_equivalence.py tests/regression/test_golden_master.py -v\n\n```json:metadata\n{\"files\": [\"software/src/io_layer.py\",\"software/src/pipeline.py\",\"software/tests/regression/test_streaming_equivalence.py\"], \"verifyCommand\": \"uv run pytest tests/regression/test_streaming_equivalence.py tests/regression/test_golden_master.py -v\", \"acceptanceCriteria\": [\"decision gate recorded with RSS measurement\",\"streaming output byte-identical to in-memory\",\"golden master still passes\",\"env-gated, argv unchanged\"], \"optional\": true}\n```" } ], - "lastUpdated": "2026-06-10" + "lastUpdated": "2026-06-10T00:00:05Z" }