Merge pull request 'feat(datagen): add reserve grid positions and batch codes to lab config' (#2) from dev_masha into master

Reviewed-on: #2
2026-07-11 14:16:32 -04:00
parent 9144d74d4c e3359383c0
commit ede48f1c60
9 changed files with 770 additions and 8 deletions
--- a/.cursorrules
+++ b/.cursorrules
@@ -4,6 +4,16 @@

 Every commit message you generate MUST follow this convention exactly.

+### Source of truth (staged diff only)
+
+Base the message ONLY on `git diff --cached` (the index / staged changes).
+Do not use unstaged working-tree diffs, untracked files, recent commit
+messages, or chat context.
+
+- Mention only paths that appear in the staged diff.
+- If the index is empty, output exactly: `No staged changes`
+- Never mention a file (e.g. README.md) that is not in the staged diff.
+
 First line: type(scope): description
 - lowercase imperative English summary, under 72 characters, no trailing period
 - type is one of (closed list): feat fix perf refactor security docs ci chore test
@@ -14,7 +24,8 @@ Scope selection by changed files:
 - rules   = docs/rules/**
 - docs    = other prose docs (README.md, CLAUDE.md, docs/examples/**)
 - config  = configuration files (.gitignore, .cursorrules, editable YAML defaults)
- datagen = lab-configuration and data-generation code (tasks 01-03)
+- datagen = lab-configuration and data-generation code (tasks 01-03:
+  make_lab_config.py, make_process_flow.py, generate_data.py)
 - convert = format converters: JSON-LD, SQLite, PostgreSQL, RDF (tasks 04-07)
 - bench   = benchmark queries, runner, extrapolation (tasks 08-10)
 - report  = reporting code and templates (task 11)
--- a/README.md
+++ b/README.md
@@ -58,8 +58,8 @@ machine (no parallel work). Every task writes a completion marker
 | Task | Script | Main output | Status |
 |---|---|---|---|
 | 01 Lab configuration | `make_lab_config.py` | `out/config/lab_config.yaml` | implemented |
-| 02 Process flow diagrams | `make_process_flow.py` | `out/report/process_flow.md`, `out/report/diagrams/` | planned |
-| 03 CSV data generation | `generate_data.py` | `out/csv/` tree + `MANIFEST.csv` | planned |
+| 02 Process flow diagrams | `make_process_flow.py` | `out/report/process_flow.md`, `out/report/diagrams/` | implemented |
+| 03 CSV data generation | `generate_data.py` | `out/csv/` tree + `MANIFEST.csv` | implemented |
 | 04 Convert to JSON-LD | `convert_json.py` | `out/json/full/`, `out/json/hybrid/dataset.jsonld` | planned |
 | 05 Convert to SQLite | `convert_sqlite.py` | `out/sqlite/tribo.db` | planned |
 | 06 Convert to PostgreSQL | `convert_postgresql.py` | live `tribo` DB + `out/pg/tribo.dump` | planned |
@@ -146,6 +146,8 @@ docs/
 out/            ALL generated artifacts (git-ignored, reproducible):
  config/ csv/ json/ sqlite/ pg/ rdf/ bench/ report/ .done/
 make_lab_config.py    task 01 entry script (one script per task, repo root)
+make_process_flow.py  task 02 entry script
+generate_data.py      task 03 entry script
 requirements.txt      closed dependency list (docs/rules/code-python-style.md)
 ```

--- a/common/init.py
+++ b/common/init.py
@@ -0,0 +1 @@
+"""Shared helpers for the pipeline tasks (see docs/rules/build-pipeline-tasks.md)."""
--- a/docs/rules/data-naming-units.md
+++ b/docs/rules/data-naming-units.md
@@ -60,6 +60,8 @@ canonical tokens (as they appear in CSV headers):
 | `_V` | volt | `discharge_voltage_V` |
 | `_W` | watt | `power_W` |
 | `_C` | degree Celsius | `temperature_C` |
+| `_deg` | degree (angle) | `angle_deg`, `pt_gun_tilt_deg` |
+| `_eV` | electronvolt | `energy_eV` |
 | `_pct` | percent (non-composition) | `rh_pct`, `cpu_util_pct` |
 | `_s` / `_mb` | seconds / megabytes (benchmark outputs) | `wall_s`, `peak_rss_mb` |

--- a/docs/rules/meta-cursor-rules.md
+++ b/docs/rules/meta-cursor-rules.md
@@ -46,6 +46,10 @@ one-click SCM generator falls back to Cursor defaults and emits unprefixed junk
   project`, `clarify rules`).
 6. **Allowed special forms** - `Merge ...` commits are allowed as-is;
   `release(scope): vX.Y.Z` is allowed for releases.
+7. **Staged-diff grounding** - base the message ONLY on `git diff --cached`;
+   mention only staged paths; if the index is empty, output `No staged
+   changes`; never infer from unstaged diffs, untracked files, prior
+   commits, or chat context.

 ## 3. Keeping the mirror in sync

--- a/docs/specs/03_csv_data_generation.md
+++ b/docs/specs/03_csv_data_generation.md
@@ -17,7 +17,7 @@ Write `generate_data.py` (tabs for indentation, pure stdlib + numpy). Stream row
 out/csv/
 ├── batches.csv                      # 4 rows: deposition params
 ├── simtra/
-│   └── simtra_profile_<batch>.csv   # 4 files, 1000 rows each: angle, energy_eV, pt_flux, au_flux
+│   └── simtra_profile_<batch>.csv   # 4 files, 1000 rows each: angle_deg, energy_eV, pt_flux, au_flux
 ├── batch_<B>/wafer_<W>/
 │   ├── wafer_info.csv
 │   └── coupon_<C>/
--- a/generate_data.py
+++ b/generate_data.py
@@ -0,0 +1,443 @@
+"""Task 03 - Simulated CSV data generation (~150 MB).
+
+Generates the canonical raw corpus under ./out/csv/ exactly as the
+instruments would write it, driven entirely by out/config/lab_config.yaml
+(volumes, physical-model coefficients, seed). Also writes
+./out/csv/MANIFEST.csv (path, rows, bytes, sha256) and the
+./out/.done/03.ok marker. Spec: docs/specs/03_csv_data_generation.md.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import logging
+import math
+import shutil
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import numpy as np
+import yaml
+
+logger = logging.getLogger(__name__)
+
+REPO_ROOT = Path(__file__).resolve().parent
+TASK_ID = "03"
+DEPENDS_ON = ["01"]
+
+SIZE_GATE_MIB = (120, 200)  # abort bounds on the generated tree, spec 03
+
+# One float format per column family (docs/rules/data-determinism.md):
+# repr drift between runs must never change file bytes. Precision of the two
+# bulk families also sets corpus size; chosen to land in the 140-170 MB
+# target band of spec 03.
+FMT_COF = "%.16g"
+FMT_POSITION = "%.14g"
+FMT_FORCE = "%.14g"
+FMT_MISC = "%.10g"  # wt%, um, nm, GPa, simtra, wear
+
+# RNG stream tags for per-entity derived generators.
+S_XRF, S_PROF, S_NANO, S_AFM, S_COF, S_LOOP, S_WEAR, S_SIMTRA = range(1, 9)
+
+
+class ValidationError(RuntimeError):
+	pass
+
+
+class CorpusGenerator:
+	def __init__(self, cfg: dict, csv_root: Path) -> None:
+		self.cfg = cfg
+		self.csv_root = csv_root
+		self.seed = cfg["seed"]
+		self.manifest: list[tuple[str, int, int, str]] = []
+		self.total_rows = 0
+		self.total_bytes = 0
+		self.counts = {"coupons": 0, "tracks": 0, "cycle_rows": 0, "loop_rows": 0, "xrf_rows": 0}
+		self.runs_by_batch = {r["batch_code"]: r for r in cfg["runs"]}
+
+	def rng(self, *tags: int) -> np.random.Generator:
+		return np.random.default_rng([self.seed, *tags])
+
+	def write_csv(self, relpath: str, header: str, lines: list[str]) -> None:
+		text = header + "\n" + "\n".join(lines) + "\n"
+		data = text.encode("ascii")
+		path = self.csv_root / relpath
+		path.parent.mkdir(parents=True, exist_ok=True)
+		path.write_bytes(data)
+		sha = hashlib.sha256(data).hexdigest()
+		self.manifest.append((relpath, len(lines), len(data), sha))
+		self.total_rows += len(lines)
+		self.total_bytes += len(data)
+
+	# --- top-level flat files ---
+
+	def gen_batches(self) -> None:
+		sched = self.cfg["schedule"]
+		start = datetime.fromisoformat(sched["deposition_start_date"])
+		lines = []
+		for i, b in enumerate(self.cfg["deposition_matrix"]):
+			date = (start + timedelta(days=i * sched["batch_interval_days"])).date().isoformat()
+			lines.append(
+				f"{b['batch_code']},{b['pt_gun_tilt_deg']},{b['au_gun_tilt_deg']},"
+				f"{b['pt_power_W']},{b['au_power_W']},{b['pt_discharge_V']},{b['au_discharge_V']},{date}"
+			)
+		self.write_csv(
+			"batches.csv",
+			"batch_code,pt_gun_tilt_deg,au_gun_tilt_deg,pt_power_W,au_power_W,pt_discharge_V,au_discharge_V,deposition_date",
+			lines,
+		)
+
+	def gen_runs(self) -> None:
+		plates = self.cfg["friction_assignment"]["plates_per_run"]
+		lines = [
+			f"{r['run_code']},{r['environment']},{r['date']},{plates},{r['operator']}"
+			for r in self.cfg["runs"]
+		]
+		self.write_csv("runs.csv", "run_code,environment,date,plates,operator", lines)
+
+	def gen_simtra(self) -> None:
+		pm = self.cfg["physical_models"]["simtra"]
+		n = self.cfg["volumes"]["simtra_rows_per_batch"]
+		us_pt = pm["surface_binding_energy_eV"]["pt"]
+		us_au = pm["surface_binding_energy_eV"]["au"]
+		for b_idx, b in enumerate(self.cfg["deposition_matrix"], 1):
+			rng = self.rng(b_idx, S_SIMTRA)
+			angle = np.linspace(pm["angle_deg_range"][0], pm["angle_deg_range"][1], n)
+			pt_flux = pm["flux_per_W"] * b["pt_power_W"] * np.cos(np.radians(angle - b["pt_gun_tilt_deg"])) ** 2
+			au_flux = pm["flux_per_W"] * b["au_power_W"] * np.cos(np.radians(angle - b["au_gun_tilt_deg"])) ** 2
+			pt_flux = np.maximum(pt_flux * (1 + rng.normal(0, 0.02, n)), 0)
+			au_flux = np.maximum(au_flux * (1 + rng.normal(0, 0.02, n)), 0)
+			# Thompson mean emission energy ~ 2 * surface binding energy,
+			# flux-weighted across the two species.
+			energy = (pt_flux * 2 * us_pt + au_flux * 2 * us_au) / (pt_flux + au_flux + 1e-12)
+			energy = energy + rng.normal(0, 0.2, n)
+			lines = [
+				f"{FMT_MISC % a},{FMT_MISC % e},{FMT_MISC % p},{FMT_MISC % g}"
+				for a, e, p, g in zip(angle, energy, pt_flux, au_flux)
+			]
+			self.write_csv(f"simtra/simtra_profile_{b['batch_code']}.csv", "angle_deg,energy_eV,pt_flux,au_flux", lines)
+
+	# --- per-coupon measurements ---
+
+	def gen_xrf(self, rel_dir: str, b_idx: int, w: int, c: int, au_nominal: float) -> float:
+		gx_n, gy_n = self.cfg["volumes"]["xrf_grid"]
+		noise_sd = self.cfg["physical_models"]["au_gradient"]["xrf_noise_sd_wtpct"]
+		rng = self.rng(b_idx, w, c, S_XRF)
+		au = au_nominal + rng.normal(0, noise_sd, gx_n * gy_n)
+		lines = []
+		i = 0
+		for gy in range(1, gy_n + 1):
+			for gx in range(1, gx_n + 1):
+				lines.append(f"{gx},{gy},{FMT_MISC % (100.0 - au[i])},{FMT_MISC % au[i]}")
+				i += 1
+		self.write_csv(f"{rel_dir}/xrf_map.csv", "grid_x,grid_y,pt_wtpct,au_wtpct", lines)
+		self.counts["xrf_rows"] += len(lines)
+		return float(au.mean())
+
+	def gen_profilometry(self, rel_dir: str, b_idx: int, w: int, c: int, row: int, col: int) -> float:
+		pm = self.cfg["physical_models"]["thickness_um"]
+		gx_n, gy_n = self.cfg["volumes"]["profilometry_grid"]
+		# Normalized squared wafer radius: corner coupon (r2=1) is thinnest.
+		r2 = (((col - 4) / 3.0) ** 2 + ((row - 4) / 3.0) ** 2) / 2.0
+		nominal = pm["center_um"] - (pm["center_um"] - pm["edge_um"]) * r2
+		rng = self.rng(b_idx, w, c, S_PROF)
+		t = nominal + rng.normal(0, pm["noise_sd_um"], gx_n * gy_n)
+		lines = []
+		i = 0
+		for gy in range(1, gy_n + 1):
+			for gx in range(1, gx_n + 1):
+				lines.append(f"{gx},{gy},{FMT_MISC % t[i]}")
+				i += 1
+		self.write_csv(f"{rel_dir}/profilometry.csv", "grid_x,grid_y,thickness_um", lines)
+		return float(t.mean())
+
+	def gen_nanoindentation(self, rel_dir: str, b_idx: int, w: int, c: int, au_mean: float) -> None:
+		pm = self.cfg["physical_models"]
+		n = self.cfg["volumes"]["nanoindentation_indents_per_coupon"]
+		grid_w = pm["nanoindentation"]["indent_grid"][0]
+		pitch = pm["nanoindentation"]["grid_pitch_um"]
+		max_load = pm["nanoindentation"]["max_load_mN"]
+		rng = self.rng(b_idx, w, c, S_NANO)
+		h = pm["hardness_GPa"]["intercept"] + pm["hardness_GPa"]["slope_per_au_wtpct"] * au_mean \
+			+ rng.normal(0, pm["hardness_GPa"]["noise_sd"], n)
+		er = pm["reduced_modulus_GPa"]["intercept"] + pm["reduced_modulus_GPa"]["slope_per_au_wtpct"] * au_mean \
+			+ rng.normal(0, pm["reduced_modulus_GPa"]["noise_sd"], n)
+		lines = [
+			f"{i + 1},{(i % grid_w) * pitch},{(i // grid_w) * pitch},{FMT_MISC % h[i]},{FMT_MISC % er[i]},{max_load}"
+			for i in range(n)
+		]
+		self.write_csv(
+			f"{rel_dir}/nanoindentation.csv",
+			"indent_id,x_um,y_um,hardness_GPa,reduced_modulus_GPa,max_load_mN",
+			lines,
+		)
+
+	def gen_afm(self, rel_dir: str, b_idx: int, w: int, c: int, coupon_code: str) -> float:
+		pm = self.cfg["physical_models"]["roughness_ra_nm"]
+		rng = self.rng(b_idx, w, c, S_AFM)
+		ra = float(np.exp(rng.normal(math.log(pm["median_nm"]), pm["sigma_log"])))
+		rq = ra * pm["rq_over_ra"]
+		self.write_csv(
+			f"{rel_dir}/afm.csv",
+			"ra_nm,rq_nm,image_file",
+			[f"{FMT_MISC % ra},{FMT_MISC % rq},images/afm/{coupon_code}.tiff"],
+		)
+		return ra
+
+	# --- per-track friction data ---
+
+	def cof_ss(self, au_mean: float, environment: str) -> float:
+		pm = self.cfg["physical_models"]["cof_steady_state"]
+		value = pm[environment]["intercept"] + pm[environment]["slope_per_au_wtpct"] * au_mean
+		return max(value, pm["floor"])  # deliberate model floor, spec 01
+
+	def gen_track(
+		self, rel_dir: str, b_idx: int, w: int, c: int, t: int,
+		track_code: str, run: dict, au_mean: float, plate: int, probe: int, square: int,
+	) -> None:
+		tc = self.cfg["test_conditions"]
+		pm = self.cfg["physical_models"]
+		v = self.cfg["volumes"]
+		sched = self.cfg["schedule"]
+		env = run["environment"]
+		load_mn = tc["normal_load_mN"]
+
+		started = datetime.fromisoformat(f"{run['date']}T{sched['run_start_time']}+00:00") + timedelta(
+			seconds=((plate - 1) * 12 + (square - 1) * 3 + (t - 1)) * sched["track_interval_s"]
+		)
+		counterface_id = f"{run['run_code']}-P{plate}-PR{probe}-B{t}"
+		self.write_csv(
+			f"{rel_dir}/track_info.csv",
+			"track_code,run_code,environment,load_mN,stroke_mm,speed_mm_s,counterface_id,started_at",
+			[
+				f"{track_code},{run['run_code']},{env},{load_mn},{FMT_MISC % tc['stroke_mm']},"
+				f"{FMT_MISC % tc['speed_mm_s']},{counterface_id},{started.strftime('%Y-%m-%dT%H:%M:%SZ')}"
+			],
+		)
+
+		cof_ss = self.cof_ss(au_mean, env)
+		cof_rng = self.rng(b_idx, w, c, t, S_COF)
+		ri = pm["cof_run_in"]
+		cof_0 = cof_rng.uniform(*ri["cof_0_range"])
+		tau = cof_rng.uniform(*ri["tau_cycles_range"])
+		n_cycles = v["cycles_per_track"]
+		cycles = np.arange(1, n_cycles + 1)
+		decay = (cof_0 - cof_ss) * np.exp(-cycles / tau)
+		cof = cof_ss + decay + cof_rng.normal(0, ri["noise_sd"], n_cycles)
+		self.write_csv(
+			f"{rel_dir}/cof_vs_cycle.csv",
+			"cycle,cof",
+			[f"{cyc},{FMT_COF % val}" for cyc, val in zip(cycles, cof)],
+		)
+		self.counts["cycle_rows"] += n_cycles
+
+		loop_rng = self.rng(b_idx, w, c, t, S_LOOP)
+		half_um = tc["stroke_mm"] * 1000.0 / 2.0
+		pts_half = v["loop_points_per_loop"] // 2
+		fwd = np.linspace(-half_um, half_um, pts_half)
+		rev = np.linspace(half_um, -half_um, pts_half)
+		force_sd = pm["friction_loop"]["force_noise_sd_mN"]
+		lines: list[str] = []
+		for k in range(1, v["loops_per_track"] + 1):
+			cyc = k * v["loop_every_n_cycles"]
+			mu = cof_ss + (cof_0 - cof_ss) * math.exp(-cyc / tau)
+			f_fwd = mu * load_mn + loop_rng.normal(0, force_sd, pts_half)
+			f_rev = -mu * load_mn + loop_rng.normal(0, force_sd, pts_half)
+			lines.extend(f"{cyc},{FMT_POSITION % p},{FMT_FORCE % f}" for p, f in zip(fwd, f_fwd))
+			lines.extend(f"{cyc},{FMT_POSITION % p},{FMT_FORCE % f}" for p, f in zip(rev, f_rev))
+		self.write_csv(f"{rel_dir}/friction_loops.csv", "cycle,position_um,friction_force_mN", lines)
+		self.counts["loop_rows"] += len(lines)
+
+		wm = pm["wear_archard"]
+		wear_rng = self.rng(b_idx, w, c, t, S_WEAR)
+		k_arch = wm["k0_mm3_per_N_m"][env] * (1 + wm["slope_per_au_wtpct"][env] * au_mean)
+		k_arch *= math.exp(wear_rng.normal(0, wm["noise_sigma_log"]))
+		k_arch = max(k_arch, wm["k_floor_mm3_per_N_m"])  # deliberate model floor, spec 01
+		sliding_m = n_cycles * 2 * tc["stroke_mm"] / 1000.0
+		volume_um3 = k_arch * (load_mn / 1000.0) * sliding_m * 1e9
+		self.write_csv(
+			f"{rel_dir}/wear.csv",
+			"wear_volume_um3,k_archard,sliding_distance_m",
+			[f"{FMT_MISC % volume_um3},{FMT_MISC % k_arch},{FMT_MISC % sliding_m}"],
+		)
+		self.counts["tracks"] += 1
+
+	# --- hierarchy walk ---
+
+	def generate(self) -> None:
+		self.gen_batches()
+		self.gen_runs()
+		self.gen_simtra()
+
+		h = self.cfg["hierarchy"]
+		fa = self.cfg["friction_assignment"]
+		ag = self.cfg["physical_models"]["au_gradient"]
+		sched = self.cfg["schedule"]
+		dep_start = datetime.fromisoformat(sched["deposition_start_date"])
+		reserve = set(fa["reserve_grid_positions"])
+		grid_w = h["coupon_grid_per_wafer"][0]
+
+		for b_idx, batch in enumerate(self.cfg["deposition_matrix"], 1):
+			b_code = batch["batch_code"]
+			run = self.runs_by_batch[b_code]
+			au_center = ag["batch_center_wtpct"][b_code]
+			dep_date = (dep_start + timedelta(days=(b_idx - 1) * sched["batch_interval_days"])).date().isoformat()
+			friction_idx = 0
+			for w in range(1, h["wafers_per_batch"] + 1):
+				wafer_code = f"{b_code}-W{w}"
+				wafer_dir = f"batch_{b_code}/wafer_W{w}"
+				self.write_csv(
+					f"{wafer_dir}/wafer_info.csv",
+					"wafer_code,batch_code,wafer_index,deposition_date,coupons,friction_coupons,reserve_coupons",
+					[
+						f"{wafer_code},{b_code},{w},{dep_date},{h['coupons_per_wafer']},"
+						f"{h['coupons_per_wafer'] - len(reserve)},{len(reserve)}"
+					],
+				)
+				for c in range(1, h["coupons_per_wafer"] + 1):
+					row = (c - 1) // grid_w + 1
+					col = (c - 1) % grid_w + 1
+					coupon_code = f"{wafer_code}-C{c:02d}"
+					rel_dir = f"{wafer_dir}/coupon_C{c:02d}"
+					au_nominal = au_center + ag["wafer_span_wtpct"] * (col - 4) / 3.0
+
+					au_mean = self.gen_xrf(rel_dir, b_idx, w, c, au_nominal)
+					thickness = self.gen_profilometry(rel_dir, b_idx, w, c, row, col)
+					self.gen_nanoindentation(rel_dir, b_idx, w, c, au_mean)
+					ra = self.gen_afm(rel_dir, b_idx, w, c, coupon_code)
+
+					if c in reserve:
+						assignment = "RESERVE,,,"
+					else:
+						plate = friction_idx // 24 + 1
+						probe = (friction_idx % 24) // 4 + 1
+						square = friction_idx % 4 + 1
+						friction_idx += 1
+						assignment = f"{run['run_code']},{plate},{probe},{square}"
+					self.write_csv(
+						f"{rel_dir}/coupon_info.csv",
+						"coupon_code,batch_code,wafer_code,grid_row,grid_col,thickness_um,ra_nm,au_wtpct_mean,run_code,plate,probe,square",
+						[
+							f"{coupon_code},{b_code},{wafer_code},{row},{col},{FMT_MISC % thickness},"
+							f"{FMT_MISC % ra},{FMT_MISC % au_mean},{assignment}"
+						],
+					)
+					self.counts["coupons"] += 1
+
+					if c not in reserve:
+						plate, probe, square = (int(x) for x in assignment.split(",")[1:])
+						for t in range(1, fa["tracks_per_friction_coupon"] + 1):
+							self.gen_track(
+								f"{rel_dir}/track_T{t}", b_idx, w, c, t,
+								f"{coupon_code}-T{t}", run, au_mean, plate, probe, square,
+							)
+				logger.debug("wafer %s done", wafer_code)
+			logger.info(
+				"batch %s done: %d files, %.1f MiB cumulative",
+				b_code, len(self.manifest), self.total_bytes / (1024 * 1024),
+			)
+
+	def validate_counts(self) -> None:
+		v = self.cfg["volumes"]
+		expected = {
+			"coupons": v["coupons_total"],
+			"tracks": v["tracks_total"],
+			"cycle_rows": v["cycle_rows_total"],
+			"loop_rows": v["loop_points_total"],
+			"xrf_rows": v["xrf_points_total"],
+		}
+		for key, exp in expected.items():
+			if self.counts[key] != exp:
+				raise ValidationError(f"count mismatch: {key}: generated {self.counts[key]} != expected {exp}")
+		logger.info("all entity counts match lab_config volumes: %s", self.counts)
+
+	def write_manifest(self) -> int:
+		lines = [f"{p},{r},{b},{s}" for p, r, b, s in self.manifest]
+		text = "path,rows,bytes,sha256\n" + "\n".join(lines) + "\n"
+		data = text.encode("ascii")
+		(self.csv_root / "MANIFEST.csv").write_bytes(data)
+		return len(data)
+
+
+def check_dependencies(out_root: Path) -> None:
+	for dep in DEPENDS_ON:
+		marker = out_root / ".done" / f"{dep}.ok"
+		if not marker.exists():
+			raise FileNotFoundError(f"dependency marker missing: {marker} - run task {dep} first")
+
+
+def write_marker(marker_path: Path, gen: CorpusGenerator, tree_bytes: int) -> None:
+	lines = [
+		f"task: '{TASK_ID}'",
+		"status: ok",
+		f"csv_root: {gen.csv_root.as_posix()}",
+		f"files: {len(gen.manifest)}",
+		f"total_rows: {gen.total_rows}",
+		f"total_bytes: {tree_bytes}",
+		f"coupons: {gen.counts['coupons']}",
+		f"tracks: {gen.counts['tracks']}",
+		f"cycle_rows: {gen.counts['cycle_rows']}",
+		f"loop_points: {gen.counts['loop_rows']}",
+		f"xrf_points: {gen.counts['xrf_rows']}",
+	]
+	marker_path.parent.mkdir(parents=True, exist_ok=True)
+	with open(marker_path, "w", encoding="utf-8", newline="\n") as fh:
+		fh.write("\n".join(lines) + "\n")
+
+
+def main() -> int:
+	parser = argparse.ArgumentParser(description="Task 03: generate the CSV corpus")
+	parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"])
+	parser.add_argument("--out-root", type=Path, default=REPO_ROOT / "out", help="artifact tree root (default: ./out)")
+	args = parser.parse_args()
+	logging.basicConfig(
+		level=args.log_level,
+		stream=sys.stderr,
+		format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
+	)
+
+	out_root: Path = args.out_root
+	csv_root = out_root / "csv"
+	marker_path = out_root / ".done" / f"{TASK_ID}.ok"
+	try:
+		check_dependencies(out_root)
+		if marker_path.exists():
+			logger.info("re-run: removing stale marker %s", marker_path)
+			marker_path.unlink()
+		if csv_root.exists():
+			logger.info("re-run: removing previous corpus %s", csv_root)
+			shutil.rmtree(csv_root)
+		with open(out_root / "config" / "lab_config.yaml", encoding="utf-8") as fh:
+			cfg = yaml.safe_load(fh)
+
+		gen = CorpusGenerator(cfg, csv_root)
+		gen.generate()
+		gen.validate_counts()
+		manifest_bytes = gen.write_manifest()
+
+		tree_bytes = gen.total_bytes + manifest_bytes
+		tree_mib = tree_bytes / (1024 * 1024)
+		print(f"generated tree size: {tree_mib:.1f} MiB ({tree_bytes} bytes, {len(gen.manifest) + 1} files)")
+		if not SIZE_GATE_MIB[0] <= tree_mib <= SIZE_GATE_MIB[1]:
+			raise ValidationError(
+				f"tree size {tree_mib:.1f} MiB outside gate {SIZE_GATE_MIB[0]}-{SIZE_GATE_MIB[1]} MiB"
+			)
+		write_marker(marker_path, gen, tree_bytes)
+	except Exception:
+		logger.critical("task %s failed", TASK_ID, exc_info=True)
+		return 1
+
+	c = gen.counts
+	print(
+		f"rows: total={gen.total_rows} cycles={c['cycle_rows']} loops={c['loop_rows']} "
+		f"xrf={c['xrf_rows']} coupons={c['coupons']} tracks={c['tracks']}"
+	)
+	print(f"manifest: {csv_root / 'MANIFEST.csv'} ({len(gen.manifest)} entries)")
+	print(f"marker: {marker_path}")
+	return 0
+
+
+if __name__ == "__main__":
+	sys.exit(main())
--- a/make_lab_config.py
+++ b/make_lab_config.py
@@ -136,6 +136,9 @@ LAB_CONFIG: dict = {
 		"tracks_per_friction_coupon": 3,
 		"counterfaces_per_holder": 3,
 		"fresh_counterface_per_track": True,
+		# QA witness pattern: rows 1/4/7 x cols 1/4/7 of the 7x7 grid stay
+		# unworn (reserve), sampling low/mid/high Au columns on every wafer.
+		"reserve_grid_positions": [1, 4, 7, 22, 25, 28, 43, 46, 49],
 	},
 	"volumes": {
 		"coupons_total": 588,
@@ -155,15 +158,17 @@ LAB_CONFIG: dict = {
 		"simtra_rows_per_batch": 1000,
 	},
 	"runs": [
-		{"run_code": "R1", "environment": "lab_air", "date": "2026-04-06", "operator": "A. Reyes"},
-		{"run_code": "R2", "environment": "lab_air", "date": "2026-04-13", "operator": "K. Patel"},
-		{"run_code": "R3", "environment": "dry_n2", "date": "2026-04-20", "operator": "A. Reyes"},
-		{"run_code": "R4", "environment": "dry_n2", "date": "2026-04-27", "operator": "M. Novak"},
+		{"run_code": "R1", "batch_code": "B721", "environment": "lab_air", "date": "2026-04-06", "operator": "A. Reyes"},
+		{"run_code": "R2", "batch_code": "B722", "environment": "lab_air", "date": "2026-04-13", "operator": "K. Patel"},
+		{"run_code": "R3", "batch_code": "B723", "environment": "dry_n2", "date": "2026-04-20", "operator": "A. Reyes"},
+		{"run_code": "R4", "batch_code": "B724", "environment": "dry_n2", "date": "2026-04-27", "operator": "M. Novak"},
 	],
 	"schedule": {
 		"deposition_start_date": "2026-03-02",
 		"batch_interval_days": 7,
 		"characterization_lag_days": 3,
+		"run_start_time": "09:00:00",
+		"track_interval_s": 2100,
 	},
 	"test_conditions": {
 		"normal_load_mN": 100,
@@ -228,6 +233,9 @@ LAB_CONFIG: dict = {
 			"k_floor_mm3_per_N_m": 1.0e-09,
 			"noise_sigma_log": 0.15,
 		},
+		"friction_loop": {
+			"force_noise_sd_mN": 0.3,
+		},
 		"nanoindentation": {
 			"max_load_mN": 10,
 			"indent_grid": [5, 5],
@@ -268,10 +276,16 @@ def validate_config(cfg: dict) -> None:
 		("profilometry_points_per_coupon", v["profilometry_points_per_coupon"], v["profilometry_grid"][0] * v["profilometry_grid"][1]),
 		("runs_count", f["runs"], len(cfg["runs"])),
 		("batches_count", h["batches"], len(cfg["deposition_matrix"])),
+		("reserve_total", f["reserve_coupons_total"], len(f["reserve_grid_positions"]) * h["wafers_per_batch"] * h["batches"]),
+		("friction_per_batch", f["coupons_per_run"], (h["coupons_per_wafer"] - len(f["reserve_grid_positions"])) * h["wafers_per_batch"]),
 	]
 	for name, declared, derived in checks:
 		if declared != derived:
 			raise ValueError(f"volume identity failed: {name}: declared {declared} != derived {derived}")
+	run_batches = [r["batch_code"] for r in cfg["runs"]]
+	matrix_batches = [b["batch_code"] for b in cfg["deposition_matrix"]]
+	if sorted(run_batches) != sorted(matrix_batches):
+		raise ValueError(f"run-batch mapping mismatch: runs cover {run_batches}, matrix has {matrix_batches}")
 	logger.info("all %d volume identities hold", len(checks))


--- a/make_process_flow.py
+++ b/make_process_flow.py
@@ -0,0 +1,285 @@
+"""Task 02 - Process flow diagrams.
+
+Reads ./out/config/lab_config.yaml and writes the Mermaid documentation of
+the laboratory workflow: ./out/report/process_flow.md plus one standalone
+./out/report/diagrams/D<n>.mermaid file per diagram (D1-D6), and the
+./out/.done/02.ok completion marker. Spec: docs/specs/02_process_flow_diagrams.md.
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import sys
+from pathlib import Path
+
+import yaml
+
+logger = logging.getLogger(__name__)
+
+REPO_ROOT = Path(__file__).resolve().parent
+TASK_ID = "02"
+DEPENDS_ON = ["01"]
+
+DIAGRAM_TYPES = ("flowchart", "sequenceDiagram", "erDiagram")
+
+
+def d1_coupon_assembly(cfg: dict) -> str:
+	sub = cfg["material_system"]["substrate"]
+	dims = "x".join(str(d) for d in sub["dimensions_mm"])
+	return f"""flowchart LR
+    A([{sub['material']} Base {dims} mm]) --> B[Cleaning]
+    B --> B1([Rinsed in deionized water])
+    B1 --> B2([Sonicated in cleaning solution])
+    B --> C[Smearing Adhesive]
+    C --> C1([{cfg['material_system']['adhesion_layer']['material']} adhesive coating])
+    C --> D[Vapor Deposition System<br/>Kurt J. Lesker PVD 200]
+    D --> D1([PtAu sputter coating, gradient])
+    D --> E([Test Coupon])
+"""
+
+
+def d2_characterization_batch_assembly(cfg: dict) -> str:
+	h = cfg["hierarchy"]
+	t_lo, t_hi = cfg["material_system"]["coating"]["thickness_um_range"]
+	lines = [
+		"flowchart LR",
+		"    TC([Test Coupon]) --> S[SIMTRA simulation<br/>composition / atom energies]",
+		f"    S --> OP[Optical profilometry<br/>film thickness {t_lo}-{t_hi} um]",
+		f"    OP --> TW[Assemble Test Wafer<br/>x{h['coupons_per_wafer']} coupons]",
+		f"    TW --> TB[Assemble Test Batch<br/>x{h['wafers_per_batch']} wafers]",
+	]
+	for row in cfg["deposition_matrix"]:
+		code = row["batch_code"]
+		lines.append(
+			f"    TB --> {code}[{code}<br/>Pt {row['pt_power_W']} W / Au {row['au_power_W']} W]"
+		)
+	return "\n".join(lines) + "\n"
+
+
+def d3_testing_tree(cfg: dict) -> str:
+	v = cfg["volumes"]
+	xrf = "x".join(str(g) for g in v["xrf_grid"])
+	return f"""flowchart TD
+    TC([Test Coupon]) --> F[Friction<br/>RAPID 6-probe tribometer]
+    F --> FA([Lab Air -> COF dataset mu_normal])
+    F --> FN([Dry N2 -> COF dataset mu_dry_nit])
+    TC --> NI[Nanoindentation<br/>Bruker TI980]
+    NI --> NIR([Hardness, reduced modulus<br/>{v['nanoindentation_indents_per_coupon']} indents per coupon])
+    TC --> AF[AFM]
+    AF --> AFR([Topography -> Ra / Rq])
+    TC --> XR[micro-XRF<br/>Bruker M4 Tornado]
+    XR --> XRR([Pt/Au wt% map, {xrf} grid])
+"""
+
+
+def d4_tribometer_session(cfg: dict) -> str:
+	f = cfg["friction_assignment"]
+	return f"""sequenceDiagram
+    actor Op as Operator
+    participant Prep as Sample Prep
+    participant Plan as Excel Test Plan
+    participant SW as Control Software
+    participant TR as RAPID Tribometer
+
+    Op->>Prep: Request test
+    Op->>Prep: Get samples
+    Op->>Prep: Load samples onto plates
+    Op->>Prep: Load ball holders ({f['counterfaces_per_holder']} counterfaces per holder)
+    Op->>Plan: Create test plan<br/>(Sample Plate, Plate Location, Sample ID, Save Location,<br/>Folder Name, X/Y/Z Offset, Load, Iterations)
+    Op->>TR: Transfer plates to tribometer
+    Op->>TR: Set up platter
+    Op->>SW: Activate software
+    Op->>SW: Load plan
+    SW->>TR: Activate tribometer
+    TR-->>SW: Run reciprocating tests (execution loop, see D5)
+    Op->>SW: Software stop
+    SW-->>Op: Save avg files
+    Op->>TR: Remove plates and platters
+    Op->>SW: Close software
+    Op->>TR: Pull equipment out
+"""
+
+
+def d5_execution_loop(cfg: dict) -> str:
+	f = cfg["friction_assignment"]
+	tracks_per_run = f["coupons_per_run"] * f["tracks_per_friction_coupon"]
+	return f"""flowchart TD
+    L[Load all counterfaces] --> P{{{f['plates_per_run']} plates}}
+    P -->|for each plate| PR{{{f['probes_per_plate']} probes in parallel}}
+    PR -->|for each probe| C{{{f['coupons_per_probe_square']} coupons on coupon square}}
+    C -->|for each coupon| T[Draw track]
+    T --> RC[Rotate counterface - fresh ball per track]
+    RC -->|{f['tracks_per_friction_coupon']} tracks per coupon| T
+    T --> DONE([{f['coupons_per_run']} coupons x {f['tracks_per_friction_coupon']} tracks = {tracks_per_run} tracks per run])
+"""
+
+
+def d6_data_hierarchy(cfg: dict) -> str:
+	return """erDiagram
+    DEPOSITION_RUN ||--|| SIMTRA_PROFILE : produces
+    DEPOSITION_RUN ||--|| BATCH : creates
+    BATCH ||--|{ WAFER : contains
+    WAFER ||--|{ COUPON : contains
+    COUPON ||--|| XRF_MAP : has
+    COUPON ||--|| NANOINDENTATION : has
+    COUPON ||--|| AFM : has
+    COUPON ||--|| PROFILOMETRY : has
+    COUPON ||--o{ TRACK : "friction coupons only"
+    TRACK ||--|{ CYCLE : has
+    TRACK ||--|{ LOOP_POINT : has
+    TRACK ||--|| WEAR : has
+"""
+
+
+def deposition_table(cfg: dict) -> str:
+	lines = [
+		"| Batch | Pt:Au gun tilt | Pt power W | Au power W | Pt discharge V | Au discharge V |",
+		"|---|---|---|---|---|---|",
+	]
+	for r in cfg["deposition_matrix"]:
+		lines.append(
+			f"| {r['batch_code']} | {r['pt_gun_tilt_deg']}:{r['au_gun_tilt_deg']} deg "
+			f"| {r['pt_power_W']} | {r['au_power_W']} | {r['pt_discharge_V']} | {r['au_discharge_V']} |"
+		)
+	return "\n".join(lines) + "\n"
+
+
+def build_diagrams(cfg: dict) -> list[tuple[str, str, str, str]]:
+	"""Return (file_stem, title, description, mermaid_text) per diagram."""
+	h = cfg["hierarchy"]
+	return [
+		(
+			"D1",
+			"Coupon Assembly",
+			"Substrate preparation, adhesion layer, and gradient Pt-Au sputter deposition.",
+			d1_coupon_assembly(cfg),
+		),
+		(
+			"D2",
+			"Characterization and Batch Assembly",
+			f"Per-coupon simulation and thickness characterization, then assembly into wafers "
+			f"(x{h['coupons_per_wafer']} coupons) and batches (x{h['wafers_per_batch']} wafers), "
+			f"deposited per the parameter table below.",
+			d2_characterization_batch_assembly(cfg),
+		),
+		(
+			"D3",
+			"Testing Tree",
+			"The four characterization/testing paths every coupon can take.",
+			d3_testing_tree(cfg),
+		),
+		(
+			"D4",
+			"Tribometer Session Sequence",
+			"Operator workflow for one RAPID tribometer session, from test request to teardown.",
+			d4_tribometer_session(cfg),
+		),
+		(
+			"D5",
+			"Execution Loop",
+			"Nested iteration executed by the tribometer within one run.",
+			d5_execution_loop(cfg),
+		),
+		(
+			"D6",
+			"Data Hierarchy",
+			"Entity containment and per-entity measurement datasets (ERD-style).",
+			d6_data_hierarchy(cfg),
+		),
+	]
+
+
+def check_dependencies(out_root: Path) -> None:
+	for dep in DEPENDS_ON:
+		marker = out_root / ".done" / f"{dep}.ok"
+		if not marker.exists():
+			raise FileNotFoundError(f"dependency marker missing: {marker} - run task {dep} first")
+
+
+def validate_diagram(stem: str, text: str) -> None:
+	first = text.splitlines()[0].strip() if text.strip() else ""
+	if not first.startswith(DIAGRAM_TYPES):
+		raise ValueError(f"{stem}: unexpected diagram header {first!r}")
+
+
+def write_outputs(cfg: dict, out_root: Path) -> tuple[Path, int, int]:
+	diagrams_dir = out_root / "report" / "diagrams"
+	diagrams_dir.mkdir(parents=True, exist_ok=True)
+	diagrams = build_diagrams(cfg)
+
+	md_parts = [
+		"# Process Flow - Simulated Tribology Laboratory\n",
+		"Generated by make_process_flow.py (task 02) from out/config/lab_config.yaml.",
+		"Standalone diagram sources: ./diagrams/D1..D6.mermaid.\n",
+	]
+	total_diagram_bytes = 0
+	for stem, title, description, text in diagrams:
+		validate_diagram(stem, text)
+		path = diagrams_dir / f"{stem}.mermaid"
+		with open(path, "w", encoding="utf-8", newline="\n") as fh:
+			fh.write(text)
+		total_diagram_bytes += len(text.encode("utf-8"))
+		logger.info("wrote %s (%s)", path, title)
+		md_parts.append(f"## {stem} - {title}\n")
+		md_parts.append(description + "\n")
+		md_parts.append(f"```mermaid\n{text}```\n")
+		if stem == "D2":
+			md_parts.append("Deposition matrix:\n")
+			md_parts.append(deposition_table(cfg))
+
+	md_text = "\n".join(md_parts)
+	md_path = out_root / "report" / "process_flow.md"
+	with open(md_path, "w", encoding="utf-8", newline="\n") as fh:
+		fh.write(md_text)
+	return md_path, len(md_text.encode("utf-8")), total_diagram_bytes
+
+
+def write_marker(marker_path: Path, md_path: Path, md_bytes: int, diagram_bytes: int) -> None:
+	lines = [
+		f"task: '{TASK_ID}'",
+		"status: ok",
+		f"process_flow_file: {md_path.as_posix()}",
+		f"process_flow_bytes: {md_bytes}",
+		"diagrams: 6",
+		f"diagram_bytes: {diagram_bytes}",
+	]
+	marker_path.parent.mkdir(parents=True, exist_ok=True)
+	with open(marker_path, "w", encoding="utf-8", newline="\n") as fh:
+		fh.write("\n".join(lines) + "\n")
+
+
+def main() -> int:
+	parser = argparse.ArgumentParser(description="Task 02: write process flow diagrams")
+	parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"])
+	parser.add_argument("--out-root", type=Path, default=REPO_ROOT / "out", help="artifact tree root (default: ./out)")
+	args = parser.parse_args()
+	logging.basicConfig(
+		level=args.log_level,
+		stream=sys.stderr,
+		format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
+	)
+
+	out_root: Path = args.out_root
+	marker_path = out_root / ".done" / f"{TASK_ID}.ok"
+	try:
+		check_dependencies(out_root)
+		if marker_path.exists():
+			logger.info("re-run: removing stale marker %s", marker_path)
+			marker_path.unlink()
+		with open(out_root / "config" / "lab_config.yaml", encoding="utf-8") as fh:
+			cfg = yaml.safe_load(fh)
+		md_path, md_bytes, diagram_bytes = write_outputs(cfg, out_root)
+		write_marker(marker_path, md_path, md_bytes, diagram_bytes)
+	except Exception:
+		logger.critical("task %s failed", TASK_ID, exc_info=True)
+		return 1
+
+	print(f"task 02 ok: {md_path} ({md_bytes} bytes)")
+	print(f"diagrams: 6 files under {out_root / 'report' / 'diagrams'} ({diagram_bytes} bytes)")
+	print(f"marker: {marker_path}")
+	return 0
+
+
+if __name__ == "__main__":
+	sys.exit(main())
				`@@ -0,0 +1 @@`
				`"""Shared helpers for the pipeline tasks (see docs/rules/build-pipeline-tasks.md)."""`