Bound long TAS diagnostic storage without inferring guest failure

(cherry picked from commit 0b178bd84ba8319d33987cfe5a943dea538389ed)

tools/tasreplays/pepsiman.py

@@ -313,6 +313,7 @@ def run(args):
     argv=[sys.executable,HERE/'run_native.py',args.output,'--exe',info['executable'],
           '--game',info['game'],'--disc',info['disc'],'--bios',info['bios'],'--route',info['route'],
           '--cd-source-clock-tape',info['tape'],'--neutral-tail',str(observation_end-FRAMES+1),'--timeout',str(args.timeout),
+          '--storage-budget-mib','1536',
           '--checkpoint-every','1200','--renderer','software',*tekken3.PROFILE]
     # The same native implementations are an explicit candidate: all psx/
     # source bytes match 2.2.2 -> 2.3, but this title still needs its own gates.
@@ -344,6 +345,11 @@ def run(args):
             report.update(status='fail',error=str(error))
     elif process.returncode:
         report['status']='fail'
+    if (args.output/'exit.json').exists():
+        host=json.loads((args.output/'exit.json').read_text())
+        if host.get('timed_out') or host.get('stop_reason') in {'host_timeout','host_storage_budget'}:
+            report.update(status='incomplete',host_limit=host.get('stop_reason') or 'host_timeout',
+                          classification='host resource limit; no guest failure inferred')
     write(args.output/'verification.json',report)
     print(json.dumps(report))
     return 0 if report['status']=='pass' else 2 if report['status']=='incomplete' else 1

tools/tasreplays/process_budget.py

@@ -0,0 +1,30 @@
+"""Host resource limits for retained diagnostics; never infer a guest verdict."""
+import subprocess
+import time
+
+
+def wait_budgeted(process, directory, timeout, max_bytes=None, max_files=1000, interval=2):
+    started=time.monotonic()
+    reason=None
+    inventory={'bytes':0,'files':0}
+    while True:
+        exited=process.poll() is not None
+        files=[p for p in directory.rglob('*') if p.is_file()]
+        sizes=[]
+        for p in files:
+            try:sizes.append(p.stat().st_size)
+            except FileNotFoundError:pass
+        inventory={'bytes':sum(sizes),'files':len(sizes)}
+        if max_bytes is not None and (inventory['bytes']>max_bytes or inventory['files']>max_files):reason='host_storage_budget'
+        elif not exited and time.monotonic()-started>=timeout:reason='host_timeout'
+        if reason:
+            if not exited:
+                process.terminate()
+                try:process.wait(timeout=15)
+                except subprocess.TimeoutExpired:process.kill();process.wait(timeout=15)
+            break
+        if exited:break
+        try:process.wait(timeout=min(interval,max(0.01,timeout-(time.monotonic()-started))))
+        except subprocess.TimeoutExpired:pass
+    return {'exit_code':process.returncode,'stop_reason':reason,'last_inventory':inventory,
+            'host_seconds':time.monotonic()-started}

tools/tasreplays/run_native.py

@@ -86,6 +86,7 @@ def main():
     parser.add_argument("--update-predictor",choices=("gate", "previous-accept"),default="gate",
                         help="experimental prediction only; actual packet/context acceptance stays authoritative")
     parser.add_argument("--checkpoint-every", type=int, default=300)
+    parser.add_argument('--storage-budget-mib',type=int,help='Stop with host_storage_budget if diagnostic output exceeds this bound')
     parser.add_argument("--watch-u16", type=lambda x: int(x, 0), action="append", default=[],
                         help="read a physical RAM u16 before each next input (max32)")
     parser.add_argument("--record-frame", type=int,
@@ -165,6 +166,8 @@ def main():
         raise ValueError("neutral tail outside 0..60000 ticks")
     if not 1 <= args.checkpoint_every <= 10000:
         raise ValueError("checkpoint interval outside 1..10000")
+    if args.storage_budget_mib is not None and not 1<=args.storage_budget_mib<=16384:
+        raise ValueError('storage budget must be in 1..16384 MiB')
     if len(args.watch_u16) > 32 or len(set(args.watch_u16)) != len(args.watch_u16) or any(
             a < 0 or a > 0x1ffffe or a % 2 for a in args.watch_u16):
         raise ValueError("watch addresses must be unique aligned physical RAM u16 offsets (max32)")
@@ -378,7 +381,8 @@ p2_mode = "digital"
         "launcher_sha256": digest(Path(__file__)),
         "boundary": "N records supplied before the next normal VBlank sample",
         "guest_controller_polls_measured": False,
-        "source_timing_equivalence": "unqualified", "timeout_seconds": args.timeout})
+        "source_timing_equivalence": "unqualified", "timeout_seconds": args.timeout,
+        "storage_budget_mib":args.storage_budget_mib})
     startup = None
     if os.name == "nt":
         startup = subprocess.STARTUPINFO()
@@ -391,19 +395,19 @@ p2_mode = "digital"
                                    stdout=out, stderr=err)
         write_json(run / "process.json", {"pid": process.pid})
         print(f"Native PID {process.pid}; evidence: {run}", flush=True)
-        try:
-            code = process.wait(timeout=args.timeout)
-        except subprocess.TimeoutExpired:
-            timed_out = True
-            process.terminate()
-            code = process.wait(timeout=15)
+        from process_budget import wait_budgeted
+        budget=wait_budgeted(process,run,args.timeout,
+                             args.storage_budget_mib*1024**2 if args.storage_budget_mib is not None else None)
+        code=budget['exit_code']
+        timed_out=budget['stop_reason']=='host_timeout'
     write_json(run / "exit.json", {"pid": process.pid, "exit_code": code,
+               "stop_reason":budget['stop_reason'],"diagnostic_inventory":budget['last_inventory'],
                "timed_out": timed_out, "host_seconds": time.monotonic() - start,
                "completion_exists": (run / "complete.json").exists()})
     complete = None
     if (run / "complete.json").exists():
         complete = json.loads((run / "complete.json").read_text())
-    qualified = (not timed_out and code == 0 and complete is not None
+    qualified = (budget['stop_reason'] is None and code == 0 and complete is not None
                  and complete["frame"] == identity["frames"] + args.neutral_tail
                  and complete["input_frames"] == identity["frames"]
                  and complete["neutral_tail_ticks"] == args.neutral_tail
@@ -411,7 +415,7 @@ p2_mode = "digital"
     if update_profile:
         endpath=run/'update-input-end.json'
         end=json.loads(endpath.read_text()) if endpath.exists() else None
-        qualified=(not timed_out and code==0 and complete is not None and end is not None
+        qualified=(budget['stop_reason'] is None and code==0 and complete is not None and end is not None
                    and end['accepted_samples']==identity['frames']
                    and end['accepted_words_sha256']==identity['words_sha256']
                    and complete['frame']==end['frame']+args.neutral_tail

tools/tasreplays/test_pepsiman.py

@@ -4,11 +4,15 @@ import hashlib
 import json
 from pathlib import Path
 import tempfile
+import os
+import subprocess
+import sys
 from unittest.mock import patch
 import pepsiman
 from observation_evidence import compare_returns, terminal_consistency, compare_stock_observations, captured_frames
 from verify_scph5500_seeds import derive
 from compare_ram_pages import MAGIC, page_hash
+from process_budget import wait_budgeted
 
 def cue():
     parts=[]
@@ -149,4 +153,34 @@ class Observations(unittest.TestCase):
         (observed/'frame-000002.png').unlink()
         with self.assertRaisesRegex(ValueError,'inventory'):compare()
 
+
+class HostBudgets(unittest.TestCase):
+    def start(self,root,program):
+        return subprocess.Popen([sys.executable,'-c',program,str(root)],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,
+                                creationflags=subprocess.CREATE_NO_WINDOW if os.name=='nt' else 0)
+
+    def test_storage_budget_stops_active_process(self):
+        with tempfile.TemporaryDirectory() as temp:
+            root=Path(temp)
+            p=self.start(root,"import pathlib,sys,time;pathlib.Path(sys.argv[1],'capture').write_bytes(b'x'*4096);time.sleep(10)")
+            result=wait_budgeted(p,root,3,max_bytes=1024,interval=0.02)
+            self.assertEqual(result['stop_reason'],'host_storage_budget')
+            self.assertIsNotNone(p.poll())
+
+    def test_terminal_storage_budget_and_timeout(self):
+        with tempfile.TemporaryDirectory() as temp:
+            root=Path(temp)
+            p=self.start(root,"import pathlib,sys;pathlib.Path(sys.argv[1],'capture').write_bytes(b'x'*4096)")
+            p.wait(timeout=5)
+            self.assertEqual(wait_budgeted(p,root,3,max_bytes=1024)['stop_reason'],'host_storage_budget')
+            p=self.start(root,'import time;time.sleep(10)')
+            self.assertEqual(wait_budgeted(p,root,0.1,interval=0.02)['stop_reason'],'host_timeout')
+
+    def test_complete_process_has_no_guest_verdict(self):
+        with tempfile.TemporaryDirectory() as temp:
+            p=self.start(Path(temp),'pass')
+            result=wait_budgeted(p,Path(temp),3,max_bytes=1024,interval=0.02)
+            self.assertEqual(result['exit_code'],0)
+            self.assertIsNone(result['stop_reason'])
+
 if __name__=='__main__': unittest.main()