test: parameterize matched public MRT Jet diagnostics

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.

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)

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"):
+def plan(h, collision="SRT", observation_mode="last_base_step", jet_scenario="none"):
-    _, _, c = config(h, 40, collision)
+    _, _, c = config(h, 40, collision, jet_scenario)
-    _, _, f = config(h, 80, collision)
+    _, _, 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)
+    cfgdict, bodydict, p = config(h, 40 if refined else 80, a.collision, a.jet_scenario)
-    cap = 5.0 if a.mode == "pilot" else 300.0
+    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,
+        gates=periodic.GATES if a.jet_scenario in ("none", "zero") else None,
-        reference_gates=periodic.REFERENCE_GATES,
+        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
     )