Fix replay of helper allocations above the single-width limit

docs/research/helper-lifetime-allocation.md

@@ -46,6 +46,12 @@ Limits: 512 lifetimes, endpoints in `0..10^12`, widths in `1..10^60`, 1 MiB inpu
 and 32 MiB replay certificate. Invalid input or a false certificate exits 2
 without a success report. Exact input bytes are bound by SHA-256.
 
+Certificate capacities and slot endpoints can reach `512 * 10^60`, the sum
+of all allowed widths. The certificate reader accepts that aggregate bound;
+the lifetime input keeps its `10^60` width limit. For example, two simultaneous
+helpers of width `10^60` produce a `2 * 10^60` certificate that can be replayed
+through `--verify`. Oversized integer tokens remain bounded before conversion.
+
 ## Frozen P104 example
 
 `tests/fixtures/helper-lifetimes-pr5717.json` copies the eleven rows in
@@ -66,3 +72,5 @@ source-specific evidence or certify a new κ.
 Credits: classical interval coloring, the T74 cleanup pool and lower-bound
 witness in PR2655, the P104 owned-interface audit in PR5717, and their retained
 predecessors. Prepared for Ali Romman with OpenAI Codex assistance. Apache-2.0.
+Aggregate-capacity replay repair and CLI regression controls by
+josecarlosgonzalezroque-design with substantial OpenAI Codex assistance.

scripts/allocate_helper_lifetimes.py

@@ -22,6 +22,17 @@
 
 MAX_LIVES=512
 MAX_WIDTH=10**60
+MAX_CAPACITY=MAX_LIVES*MAX_WIDTH
+
+
+def _certificate_integer(text):
+    # Aggregate pool coordinates can exceed one lifetime's width limit.
+    # Bound the token before int(), independently of Python's digit limit.
+    require(len(text.removeprefix('-'))<=len(str(MAX_CAPACITY)),
+            'certificate integer token exceeds pool capacity digit limit')
+    value=int(text)
+    require(abs(value)<=MAX_CAPACITY,'certificate integer exceeds maximum pool capacity')
+    return value
 
 
 def validate(problem):
@@ -75,7 +86,7 @@ def verify(problem,certificate):
     require(type(certificate) is dict and certificate.keys()=={'schema','capacity','allocations','peak'} and
             certificate['schema']=='integer-mult-helper-allocation/v1','invalid allocation schema')
     capacity=certificate['capacity'];allocations=certificate['allocations'];peak=certificate['peak']
-    require(type(capacity) is int and 1<=capacity<=MAX_LIVES*MAX_WIDTH,'invalid pool capacity')
+    require(type(capacity) is int and 1<=capacity<=MAX_CAPACITY,'invalid pool capacity')
     require(type(allocations) is dict and allocations.keys()==by_id.keys(),'allocation IDs differ')
     pieces=0
     for name,slots in allocations.items():
@@ -113,10 +124,10 @@ def verify(problem,certificate):
                     'are declarations; no native map or kappa improvement is certified.'}
 
 
-def load(path,limit):
+def load(path,limit,*,parse_int=_integer):
     with Path(path).open('rb') as stream:raw=stream.read(limit+1)
     require(len(raw)<=limit,'input exceeds byte limit')
-    value=json.loads(raw.decode('utf-8'),object_pairs_hook=_pairs,parse_int=_integer,
+    value=json.loads(raw.decode('utf-8'),object_pairs_hook=_pairs,parse_int=parse_int,
                      parse_float=_not_integer,parse_constant=_not_integer)
     return value,{'path':str(path),'bytes':len(raw),'sha256':sha256(raw).hexdigest()}
 
@@ -127,7 +138,8 @@ def main(argv=None):
     args=parser.parse_args(argv)
     try:
         problem,pin=load(args.lifetimes,1024*1024);inputs={'lifetimes':pin}
-        if args.verify:certificate,inputs['certificate']=load(args.verify,32*1024*1024)
+        if args.verify:certificate,inputs['certificate']=load(args.verify,32*1024*1024,
+                                                            parse_int=_certificate_integer)
         else:certificate=allocate(problem)
         report=verify(problem,certificate);report['certificate']=certificate;report['inputs']=inputs
         report['schema']='integer-mult-helper-allocation-report/v1';report['contract']=problem['contract']

tests/test_research_helper_allocation.py

@@ -94,4 +94,45 @@ def test_isolated_optimized_cli_and_replay_hashes(self):
             cert.write_text('{}');r=subprocess.run(command+['--verify',str(cert)],capture_output=True,text=True)
             self.assertEqual(r.returncode,2);self.assertEqual(r.stdout,'')
 
+    def test_cli_replays_large_aggregate_capacity(self):
+        # A pool can exceed the per-lifetime width bound without any input
+        # exceeding it. Replay the actual producer's serialized certificate.
+        for count in (2,512):
+            with self.subTest(lifetimes=count), tempfile.TemporaryDirectory() as tmp:
+                source=Path(tmp)/'lifetimes.json';cert=Path(tmp)/'cert.json'
+                source.write_text(json.dumps(problem([(0,1,10**60)]*count)))
+                command=[sys.executable,'-I','-O','-B',str(TOOL),str(source)]
+                produced=subprocess.run(command,capture_output=True,text=True,check=True,cwd=tmp)
+                certificate=json.loads(produced.stdout)['certificate']
+                self.assertEqual(certificate['capacity'],count*10**60)
+                cert.write_text(json.dumps(certificate))
+                replay=subprocess.run(command+['--verify',str(cert)],capture_output=True,
+                                      text=True,check=True,cwd=tmp)
+                report=json.loads(replay.stdout)
+                self.assertTrue(report['valid'])
+                self.assertEqual(report['certificate'],certificate)
+                self.assertEqual(report['minimum_capacity'],count*10**60)
+                self.assertEqual(report['inputs']['lifetimes']['sha256'],sha256(source.read_bytes()).hexdigest())
+                self.assertEqual(report['inputs']['certificate']['sha256'],sha256(cert.read_bytes()).hexdigest())
+
+    def test_cli_keeps_input_and_certificate_integer_bounds(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            source=Path(tmp)/'lifetimes.json';cert=Path(tmp)/'cert.json'
+            p=problem([(0,1,10**60),(0,1,10**60)])
+            source.write_text(json.dumps(p));good=M['allocate'](p)
+            command=[sys.executable,'-I','-O','-B',str(TOOL),str(source)]
+            maximum=512*10**60
+            for capacity in (2*10**60+1,maximum+1,-maximum-1,10**1000):
+                with self.subTest(capacity=capacity):
+                    bad=copy.deepcopy(good);bad['capacity']=capacity
+                    cert.write_text(json.dumps(bad))
+                    r=subprocess.run(command+['--verify',str(cert)],capture_output=True,text=True,cwd=tmp)
+                    self.assertEqual(r.returncode,2);self.assertEqual(r.stdout,'')
+                    self.assertEqual(json.loads(r.stderr)['status'],'INVALID_HELPER_ALLOCATION')
+            p['lifetimes'][0]['width']=10**60+1
+            source.write_text(json.dumps(p))
+            r=subprocess.run(command,capture_output=True,text=True,cwd=tmp)
+            self.assertEqual(r.returncode,2);self.assertEqual(r.stdout,'')
+            self.assertEqual(json.loads(r.stderr)['status'],'INVALID_HELPER_ALLOCATION')
+
 if __name__=='__main__':unittest.main()