diff --git a/docs/validation/refined_dfg_window.md b/docs/validation/refined_dfg_window.md index 1998c40..0849c62 100644 --- a/docs/validation/refined_dfg_window.md +++ b/docs/validation/refined_dfg_window.md @@ -56,3 +56,29 @@ Root must independently review and admit any execution. An admitted paired pilot the existing watchdog, device selection, `--gpu-gates-approved`, positive `--max-wall-seconds` and a new output directory, adding only `--exchange-window` to the selected collision's pilot command. No GPU or remote action was performed for this patch. + +## Public Jet diagnostic option + +`--collision MRT --exchange-window --jet-scenario zero|fixed|reverse` selects +unshifted FP32, no LES, constant transfer on both existing public backends. Default +`none` and the historical manifests remain unchanged. `zero` registers both Jets at +zero amplitude and retains every existing DFG verdict, including pressure. `fixed` +uses outward radial parabolic sectors at pi/2 and 3pi/2, half-width pi/12, peaks +(+.001,-.001). Both use the existing D40/D80 geometry and U=.01. Sector origins are +(80.5,81) in root edge coordinates and (160,160.5) in uniform node coordinates; +radial extent is [0,1.005R]. `reverse` keeps those peaks through t*=300, updates both +public setters at that completed boundary, and continues to 305 with opposite signs. + +Fixed diagnostic windows are 250–260, 260–270, 270–280, 280–290 and 290–300; reversal +adds 300–305. Histories retain exact interval endpoints and root/uniform lattice +impulses. The refined window uses total exchange; uniform uses public `read_force` +batch averages. Driven windows report mean/min/max/RMS Cd/Cl without applying DFG +reference or periodicity gates. Exit zero means healthy execution, not qualification; +`matched_case_pass` remains false. No independent CV or stress-load claim is made. + +This opt-in records current source/helper hashes with the existing manifest format, +checkout revision/dirty state and runner hash. Pilot/full must match that snapshot +and the exact scenario. It does not relabel current source as historical evidence. +Use the existing bounded plan/pilot/full commands with the three option flags above; +full requires the matching completed t*=5 paired pilot. `step_seconds` includes force +readback; observation/overhead and total wall time are separate, not kernel throughput. diff --git a/tests/unit/test_refined_dfg.py b/tests/unit/test_refined_dfg.py index df53556..8af3843 100644 --- a/tests/unit/test_refined_dfg.py +++ b/tests/unit/test_refined_dfg.py @@ -146,9 +146,9 @@ def test_full_requires_completed_pilot(): r.verify_pilot(None, r.verify_source(r.DEFAULT_SOURCE, "MRT", "window")) -def pilot_fixture(path, observation_mode="window"): +def pilot_fixture(path, observation_mode="window", jet_scenario="none"): """Synthetic contract fixture only; never a physical result or run receipt.""" - manifest = (r.verify_source(r.DEFAULT_SOURCE, "MRT", "window") + manifest = (r.verify_source(r.DEFAULT_SOURCE, "MRT", "window", jet_scenario) if observation_mode == "window" else json.loads((r.HERE / "data/refined_dfg/source_manifest_mrt.json").read_text())) (path / "plan.json").write_text( @@ -156,6 +156,7 @@ def pilot_fixture(path, observation_mode="window"): dict( source_manifest=manifest, collision="MRT", + jet_scenario=jet_scenario, observation_mode=observation_mode, bounds_root_edge=list(r.BOUNDS), runner_sha256=r.runner_hash(), @@ -190,6 +191,7 @@ def pilot_fixture(path, observation_mode="window"): dict( case=case, collision="MRT", + jet_scenario=jet_scenario, observation_mode=observation_mode, mode="pilot", status="pilot_complete", @@ -207,6 +209,7 @@ def pilot_fixture(path, observation_mode="window"): dict( case=case, collision="MRT", + jet_scenario=jet_scenario, observation_mode=observation_mode, mode="pilot", step=interval, @@ -451,6 +454,73 @@ def test_window_contract_rejection(): assert not backend.events +def test_public_jet_configuration_and_geometry(): + for diameter, center, radius in ((40, (80.5, 81), 20), (80, (160, 160.5), 40)): + cfg, _, _ = r.config(h, diameter, "MRT", "fixed") + assert not cfg["method"]["ddf_shifting"] and not cfg["method"]["les"]["enabled"] + jets = r.public_jets(center, radius, "fixed") + assert [j.amplitude for j in jets] == [.001, -.001] + assert all(j.field.inward is False and j.profile == "parabolic" for j in jets) + assert all(j.region.r_max == radius * 1.005 for j in jets) + assert all(j.amplitude == 0 for j in r.public_jets(center, radius, "zero")) + cfg, _, _, host = r.host_geometry(r.DEFAULT_SOURCE, h, "MRT", "fixed") + assert not cfg.ddf_shifting and len(host.boundary_cells) > 0 + assert r.public_jets((0, 0), 1, "none") == () + with unittest.TestCase().assertRaises(ValueError): + r.config(h, 40, "SRT", "fixed") + + +def test_public_jet_reversal_boundary_and_windows(): + from types import SimpleNamespace + + calls = [] + handle = SimpleNamespace(set_jet_amplitude=lambda i, a: calls.append((i, a))) + for diameter in (40, 80): + boundary = round(250 * diameter / r.SPEED) + assert r.jet_interval(handle, "fixed", boundary - 3, 80, diameter) == (3, None) + switch = round(300 * diameter / r.SPEED) + assert r.jet_interval(handle, "reverse", switch - 3, 80, diameter) == (3, None) + assert not calls + n, event = r.jet_interval(handle, "reverse", switch, 80, diameter) + assert n == 80 and event["step"] == switch and event["tstar"] == 300 + assert calls == [(0, -.001), (1, .001)] + calls.clear() + assert r.jet_interval(handle, "reverse", switch + 80, 80, diameter) == (80, None) + assert r.jet_interval(handle, "fixed", switch, 80, diameter) == (80, None) + assert not calls + rows = [dict(tstar=t, tstar_force=t - .5, cd=2.0, cl=-.1) for t in range(251, 306)] + windows = r.jet_diagnostics(rows, "reverse") + assert len(windows) == 6 and all(w["complete"] for w in windows) + assert all(w["cd"]["mean"] == 2 and np.isclose(w["cl"]["mean"], -.1) for w in windows) + assert windows[-1]["covered_tstar"] == 5 + with unittest.TestCase().assertRaisesRegex(ValueError, "crosses"): + r.jet_diagnostics([dict(tstar=250.5, tstar_force=250, cd=1, cl=0)], "fixed") + assert r.plan(h, "MRT", "window", "reverse")["modes"]["full_tstar"] == 305 + + +def test_public_jet_source_and_pilot_separation(): + from tempfile import TemporaryDirectory + + with TemporaryDirectory() as tmp: + path = Path(tmp) + manifest = pilot_fixture(path, jet_scenario="fixed") + assert manifest["source_capture"].startswith("current checkout") + assert len(manifest["checkout_revision"]) == 40 + assert r.verify_pilot(path, manifest, "MRT", "window", "fixed") == 3606.0 + for scenario in ("none", "zero", "reverse"): + with unittest.TestCase().assertRaisesRegex(ValueError, "Jet scenario"): + r.verify_pilot(path, manifest, "MRT", "window", scenario) + for relative in ("uniform/summary.json", "refined/ready.json"): + receipt = path / relative + record = json.loads(receipt.read_text()) + receipt.write_text(json.dumps(dict(record, jet_scenario="zero"))) + with unittest.TestCase().assertRaises(ValueError): + r.verify_pilot(path, manifest, "MRT", "window", "fixed") + receipt.write_text(json.dumps(record)) + with unittest.TestCase().assertRaises(ValueError): + r.verify_source(r.DEFAULT_SOURCE, "MRT", "last_base_step", "fixed") + + if __name__ == "__main__": suite = unittest.TestSuite( unittest.FunctionTestCase(fn) diff --git a/tests/validation/run_refined_dfg.py b/tests/validation/run_refined_dfg.py index d43902a..6f9bba7 100644 --- a/tests/validation/run_refined_dfg.py +++ b/tests/validation/run_refined_dfg.py @@ -19,6 +19,63 @@ BOUNDS = (40, 24, 320, 140) BASE_BLOCK = (2, 881) # Exact 166 x 881 domain; no geometric padding. FINE_BLOCK = (8, 8) # 240 x 568 including four fine halo cells. SPEED = 0.01 +JET_PEAK = 0.001 +JET_WINDOWS = [(t, t + 10) for t in range(250, 300, 10)] + + +def public_jets(center, radius, scenario): + """Same physical apertures; caller supplies its public geometry frame.""" + if scenario == "none": + return () + from CelerisLab.body.jets import Jet, Sector, Radial + + peak = 0.0 if scenario == "zero" else JET_PEAK + return tuple( + Jet(Sector(center, angle, math.pi / 12, radius * 1.005), + Radial(center, inward=False), "parabolic", amplitude) + for angle, amplitude in ((math.pi / 2, peak), (3 * math.pi / 2, -peak)) + ) + + +def jet_interval(handle, scenario, begin, n, diameter): + """Apply a reversal only at a completed interval boundary, before advancing.""" + if scenario != "none": + boundaries = [round(t * diameter / SPEED) for t in (250, 260, 270, 280, 290, 300, 305)] + n = min([n] + [step - begin for step in boundaries if step > begin]) + if scenario != "reverse": + return n, None + switch = round(300 * diameter / SPEED) + if begin < switch: + return min(n, switch - begin), None + if begin == switch: + handle.set_jet_amplitude(0, -JET_PEAK) + handle.set_jet_amplitude(1, JET_PEAK) + return n, dict(step=begin, tstar=300.0, amplitudes=[-JET_PEAK, JET_PEAK]) + return n, None + + +def jet_diagnostics(rows, scenario): + """Raw interval-weighted diagnostics only; no driven-flow qualification gate.""" + windows = JET_WINDOWS + ([(300, 305)] if scenario == "reverse" else []) + result = [] + for start, end in windows: + selected = [r for r in rows if start <= r["tstar_force"] < end] + if not selected: + continue + if any(2 * r["tstar_force"] - r["tstar"] < start - 1e-9 + or r["tstar"] > end + 1e-9 for r in selected): + raise ValueError("Jet diagnostic interval crosses a window boundary") + weights = np.array([2 * (r["tstar"] - r["tstar_force"]) for r in selected]) + record = dict(start_tstar=start, end_tstar=end, samples=len(selected), + covered_tstar=float(weights.sum()), + complete=bool(np.isclose(weights.sum(), end - start, rtol=0, atol=1e-9))) + for key in ("cd", "cl"): + values = np.array([r[key] for r in selected]) + record[key] = dict(mean=float(np.average(values, weights=weights)), + min=float(values.min()), max=float(values.max()), + rms=float(np.sqrt(np.average(values**2, weights=weights)))) + result.append(record) + return result def helpers(source): @@ -39,21 +96,27 @@ def helpers(source): return h, periodic -def config(h, diameter, collision="SRT"): +def config(h, diameter, collision="SRT", jet_scenario="none"): cfg, body, p = h.make_config(diameter, SPEED, collision, "neq_extrap", "zou_he_local") cfg["method"]["outlet"]["srt_neq_damp"] = 0.5 + if jet_scenario != "none": + if collision != "MRT": + raise ValueError("Public Jet scenarios require MRT") + cfg["method"]["ddf_shifting"] = False return cfg, body, p -def plan(h, collision="SRT", observation_mode="last_base_step"): - _, _, c = config(h, 40, collision) - _, _, f = config(h, 80, collision) +def plan(h, collision="SRT", observation_mode="last_base_step", jet_scenario="none"): + _, _, c = config(h, 40, collision, jet_scenario) + _, _, f = config(h, 80, collision, jet_scenario) x0, y0, x1, y1 = BOUNDS fine_active = 4 * (x1 - x0) * (y1 - y0) coarse_uncovered = c["nx"] * c["ny"] - fine_active // 4 return dict( status="prepared_not_run", collision=collision, + jet_scenario=jet_scenario, + ddf_shifting=jet_scenario == "none", observation_mode=observation_mode, root_gpu_gates_required=True, root=c, @@ -89,8 +152,18 @@ def plan(h, collision="SRT", observation_mode="last_base_step"): if collision == "SRT" else "MRT pressure NEQ, unknowns q2/6/8 extrapolated; no backflow clamp; noncanonical DFG outlet" ), - modes={"pilot_tstar": 5.0, "full_tstar": 300.0}, + modes={"pilot_tstar": 5.0, "full_tstar": 305.0 if jet_scenario == "reverse" else 300.0}, observation_dense_tstar=0.02, + **(dict(jets=dict( + uniform_origin_node=f["center"], refined_origin_root_edge=[80.5, 81.0], + angles_rad=[math.pi / 2, 3 * math.pi / 2], half_width_rad=math.pi / 12, + radial_extent_over_radius=[0.0, 1.005], + field="Radial(outward)", profile="parabolic", condition="velocity", + initial_amplitudes=[0.0, 0.0] if jet_scenario == "zero" else [JET_PEAK, -JET_PEAK], + reversal_tstar=300.0 if jet_scenario == "reverse" else None, + diagnostic_windows_tstar=JET_WINDOWS + ([(300, 305)] if jet_scenario == "reverse" else []), + transfer="constant", driven_qualification="not assessed", + )) if jet_scenario != "none" else {}), ) @@ -126,7 +199,7 @@ def initial_state(shape, origin, spacing, height=164.0, speed=SPEED, root=False) return w * (1 + 3 * cu + 4.5 * cu * cu - 1.5 * velocity * velocity) -def host_geometry(source, h, collision="SRT"): +def host_geometry(source, h, collision="SRT", jet_scenario="none"): """Production donor/transfer and complete cut-link checks, without CUDA.""" from CelerisLab.config import load_lbm_config from CelerisLab.lbm.refinement import FixedPatch2D, prepare_fine_boundary @@ -139,7 +212,7 @@ def host_geometry(source, h, collision="SRT"): from CelerisLab.body.geometry.circle import CircleGeometry from CelerisLab.body.objects import SimObject - cfgdict, _, _ = config(h, 40, collision) + cfgdict, _, _ = config(h, 40, collision, jet_scenario) with TemporaryDirectory() as tmp: p = Path(tmp) / "cfg.json" p.write_text(json.dumps(cfgdict)) @@ -153,11 +226,14 @@ def host_geometry(source, h, collision="SRT"): body = SimObject( 0, geometry=CircleGeometry(80.5, 81.0, 20.0), center=(80.5, 81.0), radius=20.0 ) - host = prepare_fine_boundary(cfg, pair, (body,), _shifted_blocked=True) + host = prepare_fine_boundary( + cfg, pair, (body,), jets=public_jets((80.5, 81.0), 20.0, jet_scenario), + _shifted_blocked=jet_scenario == "none", + ) return cfg, pair, body, host -def verify_source(source, collision="SRT", observation_mode="last_base_step"): +def verify_source(source, collision="SRT", observation_mode="last_base_step", jet_scenario="none"): if collision not in ("SRT", "MRT"): raise ValueError("Matched collision must be SRT or MRT") if observation_mode not in ("last_base_step", "window"): @@ -165,6 +241,22 @@ def verify_source(source, collision="SRT", observation_mode="last_base_step"): name = ("source_manifest_window.json" if observation_mode == "window" else "source_manifest.json" if collision == "SRT" else "source_manifest_mrt.json") expected = json.loads((HERE / "data/refined_dfg" / name).read_text()) + if jet_scenario != "none": + if collision != "MRT" or observation_mode != "window": + raise ValueError("Public Jet scenarios require MRT and --exchange-window") + import subprocess + sys.path.insert(0, str(HERE)) + from source_provenance import checkout_provenance + + paths = set(expected["sha256"]) + paths.update(subprocess.check_output( + ["git", "-C", str(source), "ls-files", "src"], text=True).splitlines()) + return dict( + source_capture="current checkout, not historical DFG qualification", + **checkout_provenance(source), observation_mode=observation_mode, + sha256={name: hashlib.sha256((source / name).read_bytes()).hexdigest() + for name in sorted(paths)}, + ) for name, digest in expected["sha256"].items(): if hashlib.sha256((source / name).read_bytes()).hexdigest() != digest: raise RuntimeError("Source mismatch: " + name) @@ -219,19 +311,20 @@ def run_case(a, h, periodic, case): from contextlib import ExitStack refined = case == "refined" - cfgdict, bodydict, p = config(h, 40 if refined else 80, a.collision) - cap = 5.0 if a.mode == "pilot" else 300.0 + cfgdict, bodydict, p = config(h, 40 if refined else 80, a.collision, a.jet_scenario) + cap = 5.0 if a.mode == "pilot" else (305.0 if a.jet_scenario == "reverse" else 300.0) max_steps = math.ceil(cap * p["diameter"] / SPEED) rows = [] healthy = True step_seconds = 0.0 + amplitude_changes = [] started = time.monotonic() check_deadline(a.deadline) dest = a.output / case dest.mkdir() with ExitStack() as stack: if refined: - cfg, pair, body, host = host_geometry(a.source, h, a.collision) + cfg, pair, body, host = host_geometry(a.source, h, a.collision, a.jet_scenario) context = stack.enter_context(CudaContext(0)) backend = stack.enter_context( FixedPatchStepper( @@ -242,6 +335,7 @@ def run_case(a, h, periodic, case): base_block=BASE_BLOCK, fine_block=FINE_BLOCK, fine_objects=(body,), + jets=public_jets((80.5, 81.0), 20.0, a.jet_scenario), observe_raw_exchange=True, **({"observe_window_exchange": True} if a.observation_mode == "window" else {}), ) @@ -274,6 +368,9 @@ def run_case(a, h, periodic, case): sim = stack.enter_context( Simulation(str(tmp / "lbm.json"), str(tmp / "body.json"), device_id=0) ) + for index, jet in enumerate(public_jets(p["center"], p["radius"], a.jet_scenario)): + if sim.add_jet(jet) != index: + raise RuntimeError("Unexpected public Jet registration index") sim.initialize() probe_p = p fluid = (sim.field.flag.reshape(p["ny"], p["nx"]) & FLUID) != 0 @@ -290,6 +387,10 @@ def run_case(a, h, periodic, case): else 0.5 ) n = min(max(1, round(width * p["diameter"] / SPEED)), max_steps - begin) + n, change = jet_interval(backend if refined else sim, a.jet_scenario, + begin, n, p["diameter"]) + if change is not None: + amplitude_changes.append(change) impulse = np.zeros(2) tick = time.monotonic() if refined: @@ -359,6 +460,10 @@ def run_case(a, h, periodic, case): / SPEED, **h.pressure_probes(rho, fluid, probe_p), ) + if a.jet_scenario != "none": + row.update(step_start=begin, tstar_start=begin * SPEED / p["diameter"], + interval_steps=n, impulse_x=float(impulse[0]), + impulse_y=float(impulse[1])) check_deadline(a.deadline) ensure_finite(row) rows.append(row) @@ -402,14 +507,21 @@ def run_case(a, h, periodic, case): steps=begin, startup_seconds=startup, step_seconds=step_seconds, + observation_and_overhead_seconds=time.monotonic() - started - startup - step_seconds, wall_seconds=time.monotonic() - started, force="raw stationary-body link momentum reaction, per-unit-depth; average over entire observation interval; normalized in each base lattice", mass_units="rho_reference * m^2 per unit depth", flux_units="rho_reference * m^2/s per unit depth", pressure="postcollision end-sample; physical pressure difference / (rho_reference * 1 m/s squared)", suite_validated=False, - **periodic.assess(rows), + **(periodic.assess(rows) if a.jet_scenario in ("none", "zero") else + dict(periodic=False, checks={}, cycles=[])), ) + if a.jet_scenario != "none": + result.update(jet_diagnostic_windows=jet_diagnostics(rows, a.jet_scenario), + amplitude_changes=amplitude_changes, + pressure_classification=("DFG reference verdict retained, including pressure" + if a.jet_scenario == "zero" else "diagnostic only; no official DFG reference gate")) if not healthy: result["periodic"] = False reference = json.loads( @@ -425,7 +537,8 @@ def run_case(a, h, periodic, case): else ( "pilot_complete" if a.mode == "pilot" - else ("periodic" if result["periodic"] else "incomplete_periodicity") + else ("driven_diagnostics_complete" if a.jet_scenario in ("fixed", "reverse") else + "periodic" if result["periodic"] else "incomplete_periodicity") ) ) (dest / "summary.json").write_text(json.dumps(result, indent=2, allow_nan=False)) @@ -464,12 +577,14 @@ def check_deadline(deadline): ) -def verify_pilot(path, manifest, collision="SRT", observation_mode="last_base_step"): +def verify_pilot(path, manifest, collision="SRT", observation_mode="last_base_step", jet_scenario="none"): if path is None: raise ValueError( "Full comparison requires --pilot-results from a completed matched pilot" ) previous = read_finite_json(path / "plan.json") + if previous.get("jet_scenario", "none") != jet_scenario: + raise ValueError("Pilot Jet scenario differs") if previous.get("observation_mode") != observation_mode: raise ValueError("Pilot observation mode differs") if previous.get("collision") != collision: @@ -484,7 +599,8 @@ def verify_pilot(path, manifest, collision="SRT", observation_mode="last_base_st for case, diameter in (("uniform", 80), ("refined", 40)): summary = read_finite_json(path / case / "summary.json") if ( - summary.get("observation_mode") != observation_mode + summary.get("jet_scenario", "none") != jet_scenario + or summary.get("observation_mode") != observation_mode or summary.get("collision") != collision or summary.get("case") != case or summary.get("mode") != "pilot" @@ -537,7 +653,8 @@ def verify_pilot(path, manifest, collision="SRT", observation_mode="last_base_st raise ValueError("Unhealthy pilot observations") ready = read_finite_json(path / case / "ready.json") if ( - ready.get("observation_mode") != observation_mode + ready.get("jet_scenario", "none") != jet_scenario + or ready.get("observation_mode") != observation_mode or ready.get("collision") != collision or ready.get("case") != case or ready.get("mode") != "pilot" @@ -555,9 +672,9 @@ def verify_pilot(path, manifest, collision="SRT", observation_mode="last_base_st or not 0 < stepping <= wall ): raise ValueError("Invalid pilot calibration timings") - # Full physical duration is 60 times pilot; dense pilot sampling makes + # Full duration is 60 (or 61 with reversal) times pilot; dense sampling makes # this conservative for the sparser early full observations. Not a guarantee. - calibrated_seconds += startup + 60 * (wall - startup) + calibrated_seconds += startup + (61 if jet_scenario == "reverse" else 60) * (wall - startup) return 3 * calibrated_seconds @@ -566,6 +683,8 @@ def main(): parser.add_argument("--exchange-window", action="store_true", help="Opt in to refined window-ledger observations on the pinned candidate") parser.add_argument("--collision", choices=["SRT", "MRT"], default="SRT") + parser.add_argument("--jet-scenario", choices=["none", "zero", "fixed", "reverse"], + default="none", help="Opt in to public unshifted MRT Jets; reverse switches at t*=300 and ends at 305") parser.add_argument("--source", type=Path, default=DEFAULT_SOURCE) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--mode", choices=["plan", "pilot", "full"], default="plan") @@ -579,13 +698,14 @@ def main(): a = parser.parse_args() a.source = a.source.resolve() a.observation_mode = "window" if a.exchange_window else "last_base_step" - manifest = verify_source(a.source, a.collision, a.observation_mode) + manifest = verify_source(a.source, a.collision, a.observation_mode, a.jet_scenario) h, periodic = helpers(a.source) from source_provenance import checkout_provenance a.provenance = checkout_provenance(a.source) + a.provenance.update(jet_scenario=a.jet_scenario, ddf_shifting=a.jet_scenario == "none") calibration = ( - verify_pilot(a.pilot_results, manifest, a.collision, a.observation_mode) + verify_pilot(a.pilot_results, manifest, a.collision, a.observation_mode, a.jet_scenario) if a.mode == "full" else None ) if a.mode != "plan" and ( @@ -602,17 +722,17 @@ def main(): if a.output.exists() and any(a.output.iterdir()): raise ValueError("Output must be empty/new") a.output.mkdir(parents=True, exist_ok=True) - specification = plan(h, a.collision, a.observation_mode) + specification = plan(h, a.collision, a.observation_mode, a.jet_scenario) specification.update( source_manifest=manifest, **a.provenance, - gates=periodic.GATES, - reference_gates=periodic.REFERENCE_GATES, + gates=periodic.GATES if a.jet_scenario in ("none", "zero") else None, + reference_gates=periodic.REFERENCE_GATES if a.jet_scenario in ("none", "zero") else None, runner_sha256=runner_hash(), wall_time_budget_seconds=budget, full_pilot_calibration_seconds=calibration, ) - host_geometry(a.source, h, a.collision) + host_geometry(a.source, h, a.collision, a.jet_scenario) (a.output / "plan.json").write_text(json.dumps(specification, indent=2)) if a.mode == "plan": return 0 @@ -650,6 +770,11 @@ def main(): status="pilot_only" if a.mode == "pilot" else "incomplete", matched_case_pass=False, ) + if a.jet_scenario in ("fixed", "reverse"): + comparison.update(status=("pilot_only" if a.mode == "pilot" else "driven_diagnostics_complete") + if len(results) == 2 and all(r["healthy"] for r in results) else "invalid_state", + qualification="not assessed; execution completion is not physical acceptance", + jet_diagnostic_windows={r["case"]: r["jet_diagnostic_windows"] for r in results}) if len(results) == 2 and all(r["healthy"] and r["periodic"] for r in results): comparison.update( status="periodic_comparison", @@ -667,7 +792,8 @@ def main(): 0 if len(results) == 2 and all(r["healthy"] for r in results) - and (a.mode == "pilot" or comparison["matched_case_pass"]) + and (a.mode == "pilot" or comparison["matched_case_pass"] + or a.jet_scenario in ("fixed", "reverse")) else 2 )