test: add cylinder-free control for bounded SRT inlet diagnosis

tests/validation/uniform_srt_diagnostic/diagnose_uniform_srt.py

@@ -1,9 +1,11 @@
 """Read-only observations of the exact failed SRT control, at most 320 steps.
 
 Default is CPU preflight. Execution requires root admission and an external watchdog.
-No solver, initialization, geometry or numerical parameter modifications.
+No solver or numerical parameter modifications. --without-cylinder removes only the body;
+all other geometry and the production initialization remain unchanged.
 """
 import argparse
+from copy import deepcopy
 import hashlib
 import importlib
 import json
@@ -47,6 +49,20 @@ def source_config(source):
     return cfg,body,p,manifest
 
 
+def select_body(body, without_cylinder=False):
+    selected = deepcopy(body)
+    if without_cylinder:
+        if len(selected["objects"]) != 1 or selected["objects"][0]["type"] != "cylinder":
+            raise ValueError("Bodyless control requires exactly the original cylinder")
+        selected["objects"] = []
+    return selected
+
+
+def body_force(sim, body):
+    # No body means no body-id 0 and no force observable; never fabricate zeros.
+    return sim.read_force(0).astype(np.float64) if body["objects"] else None
+
+
 def categories(flags,p):
     """Disjoint physical categories; subtype nibble is not a bitfield."""
     from CelerisLab.lbm.descriptors import FLUID,SOLID,OBSTACLE
@@ -172,18 +188,18 @@ def execute(a,cfg,body,p):
                         deadline_check(deadline)
                         previous=current; prior_report=report
                         sim.run(1)  # Same solver; observation buffers reset/read once per step.
-                        force=sim.read_force(0).astype(np.float64)
+                        force=body_force(sim,body)
                         current=snapshot(sim,p)
                         report=diagnose(current['g'],current['macro'],force,flags,p,current['step'])
                         if current['step']==1:
-                            print('READY '+json.dumps(dict(step=1,all_observation_paths_exercised=True,
+                            print('READY '+json.dumps(dict(**a.identity,step=1,all_observation_paths_exercised=True,
                                 physical_healthy=not report['physical_failure']),allow_nan=False),flush=True)
                         if report['physical_failure']:
                             status='first_physical_failure'; break
                         deadline_check(deadline)
             except TimeoutError:
                 status='wall_time_limit'
-            details=dict(status=status,executed_steps=int(sim.stepper.step_count),
+            details=dict(**a.identity,status=status,executed_steps=int(sim.stepper.step_count),
                 observation_step=current['step'],previous_observation_step=None if previous is None else previous['step'],
                 initial=initial_report,preceding=prior_report,current=report,
                 wall_seconds=time.monotonic()-started,
@@ -191,7 +207,7 @@ def execute(a,cfg,body,p):
                 health_scope='Physical FLUID nodes and force only; ghost/body categories cannot independently trigger physical failure.',
                 interpretation='Near-region labels locate symptoms, not the root cause. Per-step inspection changes observation cadence only. No physical acceptance.')
             size=save_states(a.output,previous,current,flags,details)
-            print(json.dumps(dict(status=status,executed_steps=int(sim.stepper.step_count),evidence_bytes=size),allow_nan=False),flush=True)
+            print(json.dumps(dict(**a.identity,status=status,executed_steps=int(sim.stepper.step_count),evidence_bytes=size),allow_nan=False),flush=True)
             return 0 if status=='step_cap_reached' else 2
 
 
@@ -200,17 +216,27 @@ def main():
     parser.add_argument('--source',type=Path,default=DEFAULT_SOURCE)
     parser.add_argument('--output',type=Path,required=True)
     parser.add_argument('--execute',action='store_true')
+    parser.add_argument('--without-cylinder',action='store_true',
+                        help='Remove only the cylinder for the bounded inlet control')
     parser.add_argument('--gpu-admitted',action='store_true')
     parser.add_argument('--wall-seconds',type=float)
     a=parser.parse_args(); a.source=a.source.resolve()
     cfg,body,p,manifest=source_config(a.source)
+    body=select_body(body,a.without_cylinder)
+    from source_provenance import checkout_provenance
+    a.identity=dict(control="without_cylinder" if a.without_cylinder else "baseline_cylinder",
+        force_observation="not_applicable_no_body" if a.without_cylinder else "per_step_body_reaction",
+        baseline_diagnostic_commit="8e378aa047bb35ae4ec46037eff7256bd2f074e7",
+        diagnostic_checkout=checkout_provenance(DEFAULT_SOURCE),
+        numerical_checkout=checkout_provenance(a.source),
+        diagnostic_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest())
     if a.output.exists() and any(a.output.iterdir()): raise ValueError('Use a new/empty output directory')
     if a.execute and (not a.gpu_admitted or not os.environ.get('CUDA_VISIBLE_DEVICES') or
             a.wall_seconds is None or not math.isfinite(a.wall_seconds) or a.wall_seconds<=0):
         raise ValueError('Explicit root admission, device selection and finite wall ceiling required')
     a.output.mkdir(parents=True,exist_ok=True)
-    dump(a.output/'preflight.json',dict(source=manifest,config=cfg,body=body,parameters=p,
-        diagnostic_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),max_steps=320,
+    dump(a.output/'preflight.json',dict(**a.identity,source=manifest,config=cfg,body=body,parameters=p,
+        max_steps=320,
         max_evidence_bytes=LIMIT_BYTES,executed=False))
     if not a.execute: return 0
     return execute(a,cfg,body,p)

tests/validation/uniform_srt_diagnostic/test_diagnostic.py

@@ -94,6 +94,38 @@ def test_source_guard_and_deadline():
     with unittest.TestCase().assertRaises(TimeoutError): d.deadline_check(0)
 
 
+def test_bodyless_selection_preserves_baseline():
+    from copy import deepcopy
+    baseline_config, baseline_body, baseline_parameters = deepcopy((cfg,body,p))
+    selected = d.select_body(body,True)
+    assert selected == {"objects": []}
+    assert (cfg,body,p) == (baseline_config,baseline_body,baseline_parameters)
+    assert d.select_body(body) == baseline_body
+    assert d.select_body(body) is not body
+    assert p['nx']==1761 and p['ny']==330 and p['mean_speed']==.01
+    assert np.isclose(3*p['viscosity']+.5,.524)
+    for invalid in ({"objects":[]},{"objects":body['objects']*2},
+                    {"objects":[{"type":"other"}]}):
+        with unittest.TestCase().assertRaises(ValueError): d.select_body(invalid,True)
+
+
+def test_absent_force_is_not_zero_or_read():
+    class FakeSimulation:
+        def __init__(self): self.calls=[]
+        def read_force(self,index):
+            self.calls.append(index)
+            return np.array([1.,2.],np.float32)
+    sim=FakeSimulation()
+    assert d.body_force(sim,{"objects":[]}) is None and not sim.calls
+    np.testing.assert_array_equal(d.body_force(sim,body),[1.,2.])
+    assert sim.calls==[0]
+    g,m,f,params=fixture()
+    report=d.diagnose(g,m,None,f,params,1)
+    assert report['force'] is None and not report['force_nonfinite']
+    m['ux'][3,1]=.2
+    assert d.diagnose(g,m,None,f,params,2)['physical_failure']
+
+
 if __name__=='__main__':
     result=unittest.TextTestRunner(verbosity=2).run(unittest.TestSuite(
         unittest.FunctionTestCase(fn) for name,fn in sorted(globals().items()) if name.startswith('test_')))