test: add opt-in fixed-window inlet population history

tests/validation/uniform_srt_diagnostic/diagnose_uniform_srt.py

@@ -21,6 +21,63 @@ import numpy as np
 HERE=Path(__file__).resolve().parent
 DEFAULT_SOURCE=HERE.parents[2]
 LIMIT_BYTES=128*1024**2
+HISTORY_LIMIT_BYTES=12*1024**2
+HISTORY_RESERVE_BYTES=64*1024
+
+
+class InletHistory:
+    """Optional copies of existing snapshots only; never changes solver/health stop."""
+    def __init__(self, flags, parameters, limit=HISTORY_LIMIT_BYTES):
+        self.flags = np.array(flags[:, :4], copy=True)
+        self.parameters = parameters
+        self.limit = limit
+        self.frames = []
+        self.steps = []
+        planned_bytes = 197 * (9 * self.flags.shape[0] * 4 * 4 + 24) + self.flags.nbytes
+        if planned_bytes + HISTORY_RESERVE_BYTES > self.limit:
+            raise ValueError('Requested history window exceeds 12 MiB budget')
+
+    def record(self, state):
+        if int(state['step']) > 196:
+            return
+        raw = state['raw']
+        if (raw.dtype != np.float32 or raw.ndim != 3 or raw.shape[0] != 9
+                or raw.shape[1] != self.flags.shape[0] or raw.shape[2] < 4):
+            raise ValueError('History requires raw shifted FP32 D2Q9 with four inlet columns')
+        step = int(state['step'])
+        if step != len(self.steps):
+            raise ValueError('History requires contiguous completed steps starting at zero')
+        frame = raw[:, :, :4]
+        self.frames.append(frame.copy())
+        self.steps.append(step)
+
+    def save(self, output, terminal_step):
+        steps = np.asarray(self.steps, dtype=np.int64)
+        raw = np.stack(self.frames) if self.frames else np.empty((0, 9, self.flags.shape[0], 4), np.float32)
+        path = output / 'inlet_history.npz'
+        np.savez(path, raw_shifted=raw, flags=self.flags, steps=steps,
+                 time_s=steps * self.parameters['dt'],
+                 tstar=steps * self.parameters['mean_speed'] / self.parameters['diameter'],
+                 x_indices=np.arange(4), y_indices=np.arange(self.flags.shape[0]),
+                 requested_step_range=np.array([0,196],np.int64),
+                 diagnostic_terminal_step=np.array(terminal_step,np.int64))
+        size = path.stat().st_size
+        if size > self.limit:
+            raise RuntimeError('History exceeded its reserved artifact bound')
+        return dict(mode='inlet_raw_history', requested_first_step=0, requested_last_step=196,
+                    complete_for_requested_window=len(steps)==197,
+                    diagnostic_terminal_step=int(terminal_step), frames=len(steps),
+                    first_step=None if not len(steps) else int(steps[0]),
+                    last_step=None if not len(steps) else int(steps[-1]),
+                    bytes=size, limit_bytes=self.limit,
+                    sha256=hashlib.sha256(path.read_bytes()).hexdigest(),
+                    layout='time,q,y,x; active postcollision raw shifted FP32; step zero is initialization',
+                    window_policy='Recording stops after step196; evolution and health stopping remain unchanged')
+
+
+def record_history(history, state):
+    if history is not None:
+        history.record(state)
 
 
 def dump(path,value):
@@ -179,6 +236,8 @@ def execute(a,cfg,body,p):
             sim.initialize()  # Exact original production initialization, unchanged.
             flags=sim.get_flags().reshape(p['ny'],p['nx'])
             current=snapshot(sim,p)
+            history=InletHistory(flags,p) if a.record_inlet_history else None
+            record_history(history,current)
             initial_report=diagnose(current['g'],current['macro'],None,flags,p,0)
             report=initial_report; status='initial_physical_failure' if report['physical_failure'] else 'step_cap_reached'
             try:
@@ -190,6 +249,7 @@ def execute(a,cfg,body,p):
                         sim.run(1)  # Same solver; observation buffers reset/read once per step.
                         force=body_force(sim,body)
                         current=snapshot(sim,p)
+                        record_history(history,current)
                         report=diagnose(current['g'],current['macro'],force,flags,p,current['step'])
                         if current['step']==1:
                             print('READY '+json.dumps(dict(**a.identity,step=1,all_observation_paths_exercised=True,
@@ -206,6 +266,8 @@ def execute(a,cfg,body,p):
                 storage='raw_shifted FP32 is exact active-buffer data; physical is production FP32 decoded G. Both are diagnostic snapshots, not complete restart checkpoints.',
                 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.')
+            if history is not None:
+                details['inlet_history']=history.save(a.output,current['step'])
             size=save_states(a.output,previous,current,flags,details)
             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
@@ -218,13 +280,18 @@ def main():
     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('--record-inlet-history',action='store_true',
+                        help='Record bounded raw x0..3 history without changing health stopping')
     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",
+    a.identity=dict(observation_mode='inlet_raw_history' if a.record_inlet_history else 'first_health_pair',
+        history_limit_bytes=HISTORY_LIMIT_BYTES if a.record_inlet_history else None,
+        history_requested_steps=[0,196] if a.record_inlet_history else None,
+        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),

tests/validation/uniform_srt_diagnostic/test_history.py

@@ -0,0 +1,91 @@
+"""CPU-only fixed-window observation tests; no GPU or simulated acceptance."""
+from contextlib import redirect_stdout
+from io import StringIO
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from types import SimpleNamespace, ModuleType
+from unittest.mock import patch
+import json
+import sys
+import unittest
+import numpy as np
+sys.path.insert(0,str(Path(__file__).resolve().parent))
+import diagnose_uniform_srt as d
+
+
+def test_raw_bits_copy_and_prefix():
+    flags=np.ones((12,20),np.uint16);p=dict(dt=.25,mean_speed=.01,diameter=80)
+    h=d.InletHistory(flags,p)
+    raw=np.arange(9*12*20,dtype=np.uint32).reshape(9,12,20).view(np.float32)
+    raw[0,0,0]=np.float32(-0.)
+    expected=raw[:,:,:4].view(np.uint32).copy()
+    h.record(dict(raw=raw,step=0)); raw.fill(7); flags.fill(0)
+    h.record(dict(raw=raw,step=1))
+    with TemporaryDirectory() as temp:
+        receipt=h.save(Path(temp),1);z=np.load(Path(temp)/'inlet_history.npz')
+        np.testing.assert_array_equal(z['raw_shifted'][0].view(np.uint32),expected)
+        assert np.all(z['flags']==1)
+        np.testing.assert_array_equal(z['steps'],[0,1])
+        np.testing.assert_array_equal(z['time_s'],[0,.25])
+        np.testing.assert_array_equal(z['tstar'],[0,.01/80])
+        assert not receipt['complete_for_requested_window'] and receipt['diagnostic_terminal_step']==1
+
+
+def test_fixed_window_bound_and_late_stop():
+    flags=np.ones((330,1761),np.uint16);p=dict(dt=.1,mean_speed=.01,diameter=80)
+    h=d.InletHistory(flags,p);raw=np.zeros((9,330,4),np.float32)
+    for step in range(321):h.record(dict(raw=raw,step=step))
+    assert len(h.frames)==197
+    with TemporaryDirectory() as temp:
+        r=h.save(Path(temp),320)
+        assert r['complete_for_requested_window'] and r['last_step']==196 and r['diagnostic_terminal_step']==320
+        assert r['bytes']<d.HISTORY_LIMIT_BYTES
+    with unittest.TestCase().assertRaises(ValueError):d.InletHistory(flags,p,limit=100)
+    with unittest.TestCase().assertRaises(ValueError):
+        d.InletHistory(flags,p).record(dict(raw=np.zeros((9,331,4),np.float32),step=0))
+
+
+def test_off_by_default():
+    import subprocess
+    with TemporaryDirectory() as temp:
+        subprocess.run([sys.executable,str(Path(d.__file__)),'--output',temp],check=True,capture_output=True)
+        r=json.loads((Path(temp)/'preflight.json').read_text())
+        assert r['observation_mode']=='first_health_pair' and r['history_requested_steps'] is None
+        assert not (Path(temp)/'inlet_history.npz').exists()
+    # Off-path must not inspect or modify observation state.
+    d.record_history(None,object())
+
+
+def test_execution_identical_and_failure_included():
+    class Sim:
+        def __init__(self,*args,**kwargs):self.stepper=SimpleNamespace(step_count=0);self.calls=[]
+        def __enter__(self):instances.append(self);return self
+        def __exit__(self,*args):pass
+        def initialize(self):self.calls.append('initialize')
+        def get_flags(self):return np.ones((12,20),np.uint16)
+        def run(self,n):self.calls.append(('run',n));self.stepper.step_count+=n
+    module=ModuleType('CelerisLab.simulation');module.Simulation=Sim
+    def snapshot(sim,p):
+        raw=np.full((9,12,20),sim.stepper.step_count,np.float32)
+        return dict(raw=raw,g=raw.copy(),macro={},step=sim.stepper.step_count)
+    def diagnose(g,m,f,flags,p,step):return dict(physical_failure=step==3)
+    p=dict(ny=12,nx=20,dt=.1,mean_speed=.01,diameter=80);instances=[];details=[]
+    def save(output,prev,current,flags,report):details.append((prev['step'],current['step'],report));return 0
+    with patch.dict(sys.modules,{'CelerisLab.simulation':module}),patch.object(d,'snapshot',snapshot),patch.object(d,'diagnose',diagnose),patch.object(d,'save_states',save),redirect_stdout(StringIO()):
+        for enabled in [False,True]:
+            with TemporaryDirectory() as temp:
+                a=SimpleNamespace(wall_seconds=100,output=Path(temp),identity={},record_inlet_history=enabled)
+                assert d.execute(a,{},dict(objects=[]),p)==2
+                if enabled:
+                    z=np.load(Path(temp)/'inlet_history.npz')
+                    np.testing.assert_array_equal(z['steps'],[0,1,2,3])
+                    assert np.all(z['raw_shifted'][-1]==3)
+                    assert details[-1][2]['inlet_history']['diagnostic_terminal_step']==3
+    assert instances[0].calls==instances[1].calls==['initialize',('run',1),('run',1),('run',1)]
+    assert [(p,c) for p,c,_ in details]==[(2,3),(2,3)]
+    assert details[0][2]['status']==details[1][2]['status']=='first_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_')))
+    raise SystemExit(not result.wasSuccessful())