csml-rpi/ai-cfd-scientist/cfd-experiment

cfd-experiment

⚠ HARD STOP — read this before any other step

If <out-dir>/checkpoints/mesh_gate_done.json does not exist (or for code_mod, if code_mod_done.json also does not exist), STOP. Running cases on the starter mesh before mesh-gate has selected the locked mesh is forbidden by Skill cfd-orchestrator's "mesh-gate before experiments" constraint. Return to Skill cfd-orchestrator and start over. The one exception: this skill is invoked recursively by cfd-mesh-gate itself with mesh-gate-role set; that path is keyed on the call coming from inside <out-dir>/mesh_gate/<group>/ and bypasses this stop.

Step 0 — Preflight gate (HARD; slice 14)

Before launching any case (sweep or single-case), run:

python scripts/stage_gate_audit.py --out-dir <out-dir> --mode preflight --target-stage experiments

If rc != 0, STOP. The script enforces that literature_done, baseline_setup_done, metric_setup_done, hypothesis_done, requirements_done, mesh_gate_done (and code_mod_done on route 2) are all on disk first. Running cases on the starter mesh before cfd-mesh-gate has selected the locked mesh is forbidden by skills/cfd-orchestrator/SKILL.md "Order constraint — mesh-gate must come BEFORE experiments". This preflight check is what enforces it at stage entry instead of after the fact.

Inside cfd-mesh-gate, the per-level runs invoke this skill with mesh-gate-role and bypass the experiments-stage preflight (mesh-gate is itself a predecessor of experiments). The bypass is keyed on the presence of <out-dir>/mesh_gate/<group>/ in the call path, not on a flag the agent can fake.

Execute one or many OpenFOAM cases. Two modes:

The wrapping concerns — pre-flight Allrun validation, runtime triage, CFL-aware retries — are agent-driven in both modes via the embedded RunValidityAgent prompts further down.

Inputs

• out-dir (required) — run dir (or mesh_gate sub-dir for mesh-gate use)
• One of:
• case_id — pulls requirement from <out-dir>/requirements.json
• requirement_text — direct requirement string (used by cfd-mesh-gate, cfd-open-discovery)
• case_dir (optional) — override the default <out-dir>/cases/<case_id>/
• mesh_policy (auto) — if <out-dir>/selected_mesh_spec.json exists for this case's group, append the locked-mesh requirement_suffix to the requirement before running
• max_loop (optional, default 10) — Foam-Agent reviewer-loop cap
• max_time_limit (optional, default 7200 seconds) — wall-clock cap for the run
• provider, model (optional) — LLM provider/model

Starter directory is READ-ONLY (per skills/cfd-orchestrator/SKILL.md non-negotiable constraints). All case mutations happen inside <out-dir>/ — sweep mode's shutil.copytree(BASE, cd) is the right pattern; never sed -i or mutate the starter source directly.

Output

• <case_dir>/ — full OpenFOAM case (mesh, fields, logs, postProcessing)
• <case_dir>/run_result.json:

``json { "status": "success|failed|timeout", "case_dir": "...", "case_name": "case_001", "case_solver": "simpleFoam", "error_logs": [], "loop_count": 1, "mesh_type": "standard_mesh|custom_mesh|gmsh", "wall_time_s": 1234.5 } ``

Sweep mode (ARIS-style — no FoamAgent handoff)

A purely scripted parameter sweep, in the shape ARIS-style frameworks use for parameter studies (typical wall-clock: tens of minutes for 30 cases vs hours for an LLM-supervised loop). The pattern is genuinely simple. Adopt it for any study where:

• A working baseline case already exists (mesh + boundary conditions + numerics validated).
• The variation across cases is parametric: changing one or a few dictionary entries (model name, coefficient, ambient condition, fluid property, etc.).
• The QoI extraction is well-known (pick the appropriate pattern in cfd-skills/cfd-stats/SKILL.md).

Do not use sweep mode when you need to generate a topology/mesh/boundary-set the framework hasn't seen — that's single-case mode (next section).

Sweep-mode pattern

For each (variant × condition):

1. Copy the baseline case to <case_dir> (cp -r baseline_case/case <case_dir>). Because custom libraries are built case-local (see cfd-code-modify — customModels/<Class>/ lives inside the case), cp -r carries the compiled .so with the copy; no global library is needed.
2. Patch the dictionary entries that vary (e.g. constant/cloudProperties.phaseChangeModel, 0/T.internalField, system/controlDict.endTime). Do not patch fields you didn't intend to change.
3. Activate any compiled custom library with a case-relative libs entry — libs ("customModels/<Class>/lib<Class>.so"); in system/controlDict — and point the relevant model selector at the new class name. Use the case-relative path, never a bare lib<Class>.so name: the bare form depends on FOAM_USER_LIBBIN / the global library search path and breaks reproducibility when the case is moved or shared.
4. Run the application named in system/controlDict (reactingFoam, simpleFoam, etc.) via Bash. Stream stdout to log.<solver>.
5. Parse the log for the QoI using a cfd-stats pattern (e.g. Pattern 1 for droplet K, Pattern 2 for Cf, Pattern 3 for drag).
6. Append one row to <out-dir>/results.csv with: variant, condition, QoI, ratio-vs-reference, sample count, t_end.

That's the entire pattern. No reviewer loop, no RAG, no LLM-graded retries.

Sweep-mode template

The agent writes this Python directly via Bash (not as a wrapper script). The template is study-agnostic — every domain-specific bit is concentrated in three small functions (configure, extract_qoi, qoi_ref) that the agent fills in for the topic at hand.

# sweep_template.py — drop into <out-dir> via Write tool, then `python sweep_template.py`
# Replace every <PLACEHOLDER> with the topic-specific value before running.
import re, csv, shutil, subprocess
from pathlib import Path
OUT  = Path("<out-dir>").resolve()
BASE = OUT / "baseline_case/case"        # known-good baseline
CASES_DIR = OUT / "cases"
CASES_DIR.mkdir(exist_ok=True)
# Define what varies. Each entry is one case to run.
# Keys are study-specific — the example below shows a model × condition sweep,
# but you can sweep any combination of dictionary entries.
SWEEP = [
    {"id": "<variant_a>_<cond_1>", "variant": "<variant_a>", "condition": <cond_1>, "coeffs": {...}},
    {"id": "<variant_a>_<cond_2>", "variant": "<variant_a>", "condition": <cond_2>, "coeffs": {...}},
    # ... fill in remaining (variant × condition) combinations
]
def configure(case_dir, spec):
    """Patch dictionaries for this spec. Read existing files, regex-substitute, write back.
    REPLACE THE BODY for the topic at hand. Pattern: locate the dictionary entry that
    encodes the swept parameter (model name, coefficient, ambient condition, fluid
    property), substitute, write back. Keep edits surgical — do NOT rewrite whole files.
    Common patches by family:
      - turbulence model     → constant/momentumTransport (or constant/turbulenceProperties)
      - viscosity / fluid    → constant/transportProperties (or constant/physicalProperties)
      - Lagrangian cloud     → constant/cloudProperties
      - radiation / thermo   → constant/<the relevant dict>
      - boundary / IC values → 0/<field>
      - numerical schemes    → system/fvSchemes / system/fvSolution / system/controlDict
    If the variant requires a compiled custom library, also add a case-relative
    `libs ("customModels/<Class>/lib<Class>.so");` entry to system/controlDict
    (idempotent — check before inserting). Never use a bare `lib<Class>.so`
    name — the case-relative path keeps the case self-contained.
    """
    raise NotImplementedError("fill in `configure` per the study's dictionary topology")
def run(case_dir):
    """Run mesh prep + the application from controlDict. Generic — same for every study."""
    if (case_dir / "system/blockMeshDict").is_file():
        subprocess.run(["blockMesh"], cwd=case_dir, check=True,
                       stdout=(case_dir / "log.blockMesh").open("w"), stderr=subprocess.STDOUT)
    # Read application name from controlDict (don't hardcode)
    app = re.search(r"application\s+(\w+);",
                    (case_dir / "system/controlDict").read_text()).group(1)
    with (case_dir / f"log.{app}").open("w") as log:
        subprocess.run([app], cwd=case_dir, check=True, stdout=log, stderr=subprocess.STDOUT)
def extract_qoi(case_dir):
    """Compute the per-case QoI from `case_dir`. Returns None if extraction fails.
    REPLACE THE BODY using the matching pattern from `cfd-skills/cfd-stats/SKILL.md`:
      Pattern 1 — droplet evaporation K (D²-law from log mass history)
      Pattern 2 — wall-shear / Cf (postProcessing/wallShearStress function-object)
      Pattern 3 — drag / lift (postProcessing/forceCoeffs function-object)
      Pattern 4 — residual plateau check
      Pattern 5 — QoI drift / settling check
      Pattern 8 — ratio metric
    Each pattern is ~30 lines. Pick one, paste here, parameterize from `case_dir`.
    """
    raise NotImplementedError("fill in `extract_qoi` per the topic's QoI family")
def qoi_ref(spec):
    """Reference value for this spec — read from a CSV / formula / DNS table.
    REPLACE THE BODY. For experiments with a closed-form correlation, return it here.
    For tabulated reference data, read once at module load and look up by spec.
    For DNS comparison, read from <out-dir>/reference_data/<file>.
    """
    raise NotImplementedError("fill in `qoi_ref` per the topic's reference correlation/data")
# --- Run the sweep (this part is generic — do not edit) ---
results = []
for spec in SWEEP:
    cd = CASES_DIR / spec["id"]
    if cd.exists(): shutil.rmtree(cd)
    shutil.copytree(BASE, cd)
    configure(cd, spec)
    try:
        run(cd)
    except subprocess.CalledProcessError:
        results.append({**spec, "qoi": None, "qoi_ref": None, "ratio": None, "error": "solver_failed"})
        continue
    qoi = extract_qoi(cd)
    if qoi is None:
        results.append({**spec, "qoi": None, "qoi_ref": None, "ratio": None, "error": "extract_failed"})
        continue
    ref = qoi_ref(spec)
    ratio = qoi / ref if ref not in (0, None) else None
    results.append({**spec, "qoi": qoi, "qoi_ref": ref, "ratio": ratio})
# Persist
with (OUT / "results.csv").open("w", newline="") as f:
    if results:
        w = csv.DictWriter(f, fieldnames=list(results[0].keys()))
        w.writeheader(); w.writerows(results)
# Summary table — mean_ratio per variant
from collections import defaultdict
by_variant = defaultdict(list)
for r in results:
    if r.get("ratio") is not None: by_variant[r["variant"]].append(r["ratio"])
print(f"{'variant':20s}  {'mean_ratio':>12s}  {'mean_abs_err':>14s}  {'N':>3s}")
for v, rs in by_variant.items():
    mr = sum(rs)/len(rs); ae = sum(abs(r-1) for r in rs)/len(rs)
    print(f"{v:20s}  {mr:12.4f}  {ae:14.4f}  {len(rs):3d}")

Run it with python <out-dir>/sweep_template.py.

Sweep-mode parallelism