Retain bounded infolog exhaustion evidence and buffered-writer controls

docs/roadmaps/barc-01-useful-play.md

@@ -164,3 +164,25 @@ engine/plugin custody and positive Replace acceptance remain open at native
 0/6. It preserves frozen browser v1 types and the old rally patch/evidence.
 Stock source merge does not activate that custom engine, qualify
 local/generated input journeys or publish/install the product.
+
+### BARC-01.5f buffered-producer continuation
+
+- [ ] **BARC-01.5f-RP1 — Reproduce stock buffering and distinguish bounded refusal causes.**
+  The source candidate adds [ordinary libc buffered-writer controls](../../tests/Broker.NativeProof/RuntimeEvidence/fixtures/complete-record-helper/test_buffered_writer.py)
+  against the actual held-FD adapter. Five public checks pass: growing partial
+  EOF exhausts 32 completed probes; a one-byte read control exhausts 128 calls
+  during the first probe; flush promotes the exact untrimmed sample; failed
+  consumption retains bounded counters with a digest receipt; destination
+  collision preserves original refusal and existing bytes. The two exhaustion
+  controls retain the same policy failure code while distinguishing their
+  mechanical cause. Two focused real compiled-policy checks pass against the
+  protected `4955ae7` artifact (4.572 seconds): both the existing writer and
+  the libc buffered writer reach exactly three completed evaluations and three
+  post-evaluation growth observations, then refuse at the unchanged cap. The
+  buffered control also retains the separate mechanical observation/receipt.
+  The [fixture contract](../../tests/Broker.NativeProof/RuntimeEvidence/fixtures/complete-record-helper/README.md)
+  describes the separate observation/receipt retained by the actual adapter in
+  the attempt root. Typed F# authority, settlement behavior and runtime caps are
+  unchanged. RP1 source delivery does not qualify useful play (still 0/6) or
+  authorize native dispatch: RP2 complete-record contract qualification and RP3
+  capacity evidence are both required before a joined successor and operation.

tests/Broker.NativeProof/RuntimeEvidence/fixtures/complete-record-helper/README.md

@@ -5,3 +5,29 @@ complete-record correspondence gate. The gate copies the Python files to a tempo
 mode-local directory and generates `source-manifest.json` from the checked-out commit and
 tree. It contains no game payload, credential, private packet, runtime grant, or accepted
 native observation. The fixture can only produce synthetic growing-log and policy inputs.
+
+
+RP1 adds `buffered_writer.py`: public atlas message bodies verified against
+Recoil `2639eedac7d1fd67d793ec93ebd27f014f336a14`, ordinary libc `fopen("w")`,
+`setvbuf(NULL, _IOFBF, 8192)` and `fprintf("%s\n")`. The libc implementation
+may choose a smaller effective buffer, exactly as stock FileSink permits. Tests
+use acknowledged writes and held writer FDs, with no fabricated transcript.
+`python3 -m unittest -v test_buffered_writer` runs the public mechanical controls
+without .NET. The real compiled policy growth controls run in the coherent gate.
+
+On a failed `consume`, the actual adapter retains
+`infolog-mechanical-<boundary>.json` and its `.receipt.json` in the validated
+attempt root, mode 0600, exclusive creation. The <=4096-byte observation contains
+only closed categories, counters, byte lengths, sample hash and monotonic elapsed
+microseconds; the receipt binds its filename, bytes/hash and source-set hash.
+No raw log, path, process identity or exception text is included. A read-call cap
+is per exact-read operation; aggregate calls across probes can exceed 128.
+`sampleSha256`, complete-prefix/tail lengths describe the last fully checked
+retained sample; requested/read-in-operation counts describe the current read.
+`settlementBudgetMicroseconds` is the frozen relative budget, or null for direct
+mechanical probes. This sidecar is separate mechanical evidence and never
+acceptance authority. Failed publication preserves the original exception and
+sets its retention flag false; absent/unjoined receipt is an evidence gap.
+The existing typed F# failure diagnostic schema, settlement semantics and caps
+remain unchanged. Source-only RP1 does not authorize another native attempt;
+RP2 contract qualification and RP3 capacity evidence remain required.

tests/Broker.NativeProof/RuntimeEvidence/fixtures/complete-record-helper/buffered_writer.py

@@ -0,0 +1,51 @@
+"""Public Recoil2639 FileSink-style libc full-buffered ordinary-file writer.
+FileSink.cpp:123-130,214-223: fopen(w), setvbuf(_IOFBF,8192), fprintf newline.
+No severity-triggered fflush, engine, content, or private inputs.
+"""
+import base64,ctypes,os,sys
+ATLAS=(b'CTextureRenderAtlas::CreateAtlasTexture()[0] atlas=IconsAtlas_0 FBO::ready=1',b'CTextureRenderAtlas::CreateAtlasTexture()[1] atlas=IconsAtlas_0 atlasRendered=0')
+ROOTS=(b'[DataDirLocater::Check] Using read-write data directory: /public/write/',b'[DataDirLocater::Check] Using read-only data directory: /public/data/',b'[DataDirLocater::Check] Isolation Mode!',b'[SpringApp::Init] Spring writeable configuration directory: /public/write/')
+class BufferedWriter:
+    def __init__(self,path):
+        self.libc=ctypes.CDLL(None)
+        self.libc.fopen.argtypes=(ctypes.c_char_p,ctypes.c_char_p);self.libc.fopen.restype=ctypes.c_void_p
+        self.libc.setvbuf.argtypes=(ctypes.c_void_p,ctypes.c_void_p,ctypes.c_int,ctypes.c_size_t)
+        self.libc.fprintf.argtypes=(ctypes.c_void_p,ctypes.c_char_p)
+        self.libc.fflush.argtypes=(ctypes.c_void_p,);self.libc.fclose.argtypes=(ctypes.c_void_p,)
+        self.stream=self.libc.fopen(os.fsencode(path),b'w');assert self.stream
+        os.chmod(path,0o600)
+        assert self.libc.setvbuf(self.stream,None,0,8192)==0 # Linux _IOFBF
+        self.path=path;self.records=0
+    def record(self,text):
+        assert b'\x00' not in text and b'\n' not in text
+        assert self.libc.fprintf(self.stream,b'%s\n',ctypes.c_char_p(text))>0
+        self.records+=1
+    def partial_eof(self):
+        previous=os.stat(self.path).st_size
+        for index in range(512): # <=64KiB public records per command
+            self.record(ATLAS[self.records%2]);size=os.stat(self.path).st_size
+            if size>previous:
+                with open(self.path,'rb') as reader:reader.seek(-1,2);last=reader.read(1)
+                if last!=b'\n':return size
+                previous=size
+        raise RuntimeError('no partial EOF in bounded public replay')
+    def close(self):self.libc.fclose(self.stream)
+def main():
+    os.umask(0o077);writer=BufferedWriter(sys.argv[1])
+    try:
+        records=ROOTS if len(sys.argv)==2 else base64.b64decode(sys.argv[2]).splitlines()
+        for record in records:writer.record(record)
+        writer.partial_eof()
+        if len(sys.argv)>2:assert writer.libc.fflush(writer.stream)==0
+        print('ready',flush=True)
+        for command in sys.stdin:
+            command=command.strip()
+            if command=='grow':writer.partial_eof()
+            elif command=='growflush':
+                writer.partial_eof();assert writer.libc.fflush(writer.stream)==0
+            elif command=='flush':assert writer.libc.fflush(writer.stream)==0
+            elif command=='close':break
+            else:raise RuntimeError('closed writer command')
+            print('done',flush=True)
+    finally:writer.close()
+if __name__=='__main__':main()

tests/Broker.NativeProof/RuntimeEvidence/fixtures/complete-record-helper/growing_log.py

@@ -1,7 +1,7 @@
 """Held-FD adapter for one source-bound, growing Recoil infolog."""
 import base64,hashlib,json,os,re,selectors,stat,subprocess,time
 from pathlib import Path
-from private_io import Refused,SHA,canonical,components,hash_artifact,need,read_bytes
+from private_io import Refused,SHA,atomic_bytes_new,atomic_new,canonical,components,hash_artifact,need,read_bytes
 from runtime_identity import mark_failure,proc_bytes,start_ticks
 MAX_LOG=4*1024*1024
 MAX_POLICY_INPUT=6*1024*1024
@@ -57,7 +57,30 @@ def _closed_state(value,status,boundary,identity,revision,sample_bytes,sample_sh
 class GrowingLog:
     def __init__(self,path,process,fd,opened,writer):
         self.path=str(path);self.process=dict(process);self.fd=fd;self.device=opened.st_dev;self.inode=opened.st_ino
-        self.previous=b'';self.revision=0;self.state=None;self.revoked=False;self.writer=writer;self.closure_sha256=None;self.failure_checkpoint='policy-scope-join';self.last_policy_observation=None;self.settlement_transcript=[];self.settlement_effects=[]
+        self.previous=b'';self.revision=0;self.state=None;self.revoked=False;self.writer=writer;self.closure_sha256=None;self.failure_checkpoint='policy-scope-join';self.last_policy_observation=None;self.settlement_transcript=[];self.settlement_effects=[];self._reset_counters(None)
+    def _reset_counters(self,boundary):
+        self.counter_started=time.monotonic();self.counter_cause=None;self.counter_deadline=None
+        self.counters={'boundary':boundary,'probesBegun':0,'probesCompleted':0,'readOperations':0,'readCalls':0,'readCallsInOperation':0,'requestedReadBytes':0,'bytesReadInOperation':0,'evaluationsBegun':0,'evaluationsCompleted':0,'growthAfterEvaluation':0,'maximumObservedBytes':0,'sampleBytes':0,'completePrefixBytes':0,'tailBytes':0}
+        prefix=self.previous.rfind(b'\n')+1
+        self.counters.update(sampleBytes=len(self.previous),completePrefixBytes=prefix,tailBytes=len(self.previous)-prefix)
+    def _counter_observation(self,check):
+        elapsed=max(0,int((time.monotonic()-self.counter_started)*1000000))
+        return {'schema':'fsbar.barc-infolog-mechanical-observation/v1',**self.counters,'elapsedMicroseconds':elapsed,'settlementBudgetMicroseconds':None if self.counter_deadline is None else max(0,int((self.counter_deadline-self.counter_started)*1000000)),'terminalCheck':check,'terminalCause':self.counter_cause or ('settlement-deadline' if check=='infolog-record-settlement-deadline' else 'other'),'sampleSha256':hashlib.sha256(self.previous).hexdigest(),'nativeAcceptance':False}
+    def _retain_counter_observation(self,error,config):
+        check=getattr(error,'_barc_failure_observation',{}).get('check',self.failure_checkpoint)
+        observation=self._counter_observation(check)
+        error._barc_infolog_mechanical_observation=observation
+        # Separate mechanical evidence: never alters the typed failure projection
+        # or acceptance. Missing publication is explicitly visible on the error.
+        try:
+            need(self.counters['boundary'] in BOUNDARIES,'closed counter boundary')
+            name='infolog-mechanical-'+self.counters['boundary']+'.json'
+            root=components(config['roots']['attemptRoot'])
+            encoded=(json.dumps(observation,separators=(',',':'),allow_nan=False)+'\n').encode()
+            atomic_bytes_new(str(root/name),encoded,4096)
+            atomic_new(str(root/(name+'.receipt.json')),{'schema':'fsbar.barc-infolog-mechanical-receipt/v1','file':name,'bytes':len(encoded),'sha256':hashlib.sha256(encoded).hexdigest(),'sourceSetSha256':_source_set(config),'nativeAcceptance':False})
+            error._barc_infolog_mechanical_retained=True
+        except Exception:error._barc_infolog_mechanical_retained=False
     @classmethod
     def acquire(cls,path,process):
         check='infolog-path';fd=None
@@ -117,12 +140,15 @@ def _deadline(deadline):
             raise mark_failure(Refused('growing log record settlement deadline'),'infolog-record-settlement-deadline','deadline',None)
     def _read_exact(self,length,deadline=None):
         chunks=[];offset=0;calls=0
+        self.counters['readOperations']+=1;self.counters['readCallsInOperation']=0;self.counters['requestedReadBytes']=length;self.counters['bytesReadInOperation']=0
         while offset<length:
             if deadline is not None:self._deadline(deadline)
             if calls>=MAX_PREAD_CALLS:
+                self.counter_cause='exact-read-call-cap'
                 raise mark_failure(Refused('growing log record exact-read exhausted'),'infolog-record-settlement-exhausted','refused',None)
+            self.counters['readCalls']+=1;self.counters['readCallsInOperation']+=1
             block=os.pread(self.fd,min(65536,length-offset),offset);need(block,'growing log truncated');chunks.append(block);offset+=len(block)
-            calls+=1
+            calls+=1;self.counters['bytesReadInOperation']=offset
             if deadline is not None:self._deadline(deadline)
         return b''.join(chunks)
     def _sample(self,deadline):
@@ -147,6 +173,7 @@ def checkpoint(name):
         need((opened.st_dev,opened.st_ino)==(self.device,self.inode),'growing log descriptor identity changed')
         checkpoint('file-custody')
         need(stat.S_ISREG(named.st_mode) and named.st_uid==os.geteuid() and stat.S_IMODE(named.st_mode)==0o600 and named.st_nlink==1,'growing log custody')
+        self.counters['maximumObservedBytes']=max(self.counters['maximumObservedBytes'],opened.st_size)
         checkpoint('size-cap');need(opened.st_size<=MAX_LOG,'growing log bound')
         writer=self._writer(self.process,path,opened,deadline)
         if expected is not None:
@@ -158,7 +185,7 @@ def _sample_impl(self,deadline):
         self._deadline(deadline)
         opened,_=self._custody(deadline=deadline)
         need(len(self.previous)<=opened.st_size,'growing log truncated')
-        length=opened.st_size;raw=self._read_exact(length,deadline)
+        length=opened.st_size;self.counters['maximumObservedBytes']=max(self.counters['maximumObservedBytes'],length);raw=self._read_exact(length,deadline)
         # This second retained-FD read closes the permanent overwrite window
         # after the first read.  The custody check also reopens no authority:
         # any failure makes this object sticky-revoked in _sample.
@@ -171,10 +198,14 @@ def _sample_impl(self,deadline):
     def _settle(self,deadline,probes):
         while probes<MAX_SETTLEMENT_PROBES:
             self._deadline(deadline)
+            self.counters['probesBegun']+=1
             raw,intact,writer=self._sample(deadline);probes+=1
+            self.counters['probesCompleted']+=1
             # Retain every fully checked byte, including an incomplete tail.
             # The next probe must extend this exact mechanical prefix.
             self.previous=raw
+            prefix=raw.rfind(b'\n')+1
+            self.counters.update(sampleBytes=len(raw),completePrefixBytes=prefix,tailBytes=len(raw)-prefix)
             complete=bool(raw) and raw.endswith(b'\n')
             event={'kind':'probe','event':'complete' if complete else 'incomplete','probe':probes,'remaining':MAX_SETTLEMENT_PROBES-probes}
             self.settlement_transcript.append({key:value for key,value in event.items() if key!='kind'})
@@ -187,6 +218,7 @@ def _settle(self,deadline,probes):
                 self._deadline(deadline)
                 time.sleep(min(SETTLEMENT_PAUSE_SECONDS,max(0,deadline-time.monotonic())))
                 self._deadline(deadline)
+        self.counter_cause='settlement-probe-cap'
         raise mark_failure(Refused('growing log record settlement exhausted'),'infolog-record-settlement-exhausted','refused',None)
     @staticmethod
     def _closure(path,digest,deadline):
@@ -298,13 +330,16 @@ def _run_policy(self,path,closure_path,closure_sha,encoded,deadline):
             if process is not None and process.stdout is not None and not process.stdout.closed:process.stdout.close()
             if process is not None and process.stderr is not None and not process.stderr.closed:process.stderr.close()
     def consume(self,boundary,config,deadline):
+        self._reset_counters(boundary)
         try:return self._consume_impl(boundary,config,deadline)
         except BaseException as error:
             self.revoked=True
-            raise mark_failure(error,self.failure_checkpoint)
+            error=mark_failure(error,self.failure_checkpoint)
+            self._retain_counter_observation(error,config)
+            raise error
     def _consume_impl(self,boundary,config,deadline):
         need(boundary in BOUNDARIES and isinstance(deadline,(int,float)),'closed growing log boundary/deadline')
-        policy_deadline=min(deadline,time.monotonic()+5)
+        policy_deadline=min(deadline,time.monotonic()+5);self.counter_deadline=policy_deadline
         policy=config['artifacts']['runtimeEvidencePolicy'];closure=config['artifacts']['runtimeEvidencePolicyClosure']
         self.failure_checkpoint='policy-artifact';hash_artifact(policy['path'],policy['sha256'],268435456)
         self.failure_checkpoint='policy-closure-precheck';closure_value=self._closure(closure['path'],closure['sha256'],policy_deadline)
@@ -325,7 +360,9 @@ def _consume_impl(self,boundary,config,deadline):
             encoded=json.dumps(request,separators=(',',':'),allow_nan=False).encode();need(len(encoded)<=MAX_POLICY_INPUT,'growing log policy input bound')
             remaining=policy_deadline-time.monotonic();need(remaining>0,'growing log policy deadline')
             self.failure_checkpoint='policy-transport';self.last_policy_observation=None
+            self.counters['evaluationsBegun']+=1
             output=self._run_policy(policy['path'],closure['path'],closure['sha256'],encoded,policy_deadline)
+            self.counters['evaluationsCompleted']+=1
             try:value=json.loads(output.decode('utf-8'),object_pairs_hook=_unique,parse_constant=lambda _:(_ for _ in ()).throw(Refused('nonfinite policy JSON')))
             except (UnicodeError,json.JSONDecodeError) as error:raise Refused('invalid growing log policy result') from error
             self.failure_checkpoint='policy-result-join'
@@ -345,7 +382,9 @@ def _consume_impl(self,boundary,config,deadline):
             hash_artifact(policy['path'],policy['sha256'],268435456)
             self._closure(closure['path'],closure['sha256'],policy_deadline)
             need(time.monotonic()<policy_deadline,'policy final closure deadline')
+            self.counters['maximumObservedBytes']=max(self.counters['maximumObservedBytes'],current.st_size)
             if current.st_size!=len(raw):
+                self.counters['growthAfterEvaluation']+=1
                 continue
             if last=='accepted':
                 self.settlement_effects.append({'kind':'terminal','outcome':'accepted','check':None,'state':self.state})
@@ -355,5 +394,7 @@ def _consume_impl(self,boundary,config,deadline):
                 raise mark_failure(Refused('growing log policy refused or unknown'),self.failure_checkpoint,'policy-nonaccepted',self.last_policy_observation)
             time.sleep(min(0.01,max(0,policy_deadline-time.monotonic())))
         self.failure_checkpoint='policy-sample-exhausted';self.revoked=True
-        if last=='accepted':raise mark_failure(Refused('growing log policy sample growth exhausted'),self.failure_checkpoint,'refused',None)
+        if last=='accepted':
+            self.counter_cause='policy-evaluation-growth-cap'
+            raise mark_failure(Refused('growing log policy sample growth exhausted'),self.failure_checkpoint,'refused',None)
         raise mark_failure(Refused(f'growing log policy {last or "unavailable"}'),self.failure_checkpoint,'policy-nonaccepted',self.last_policy_observation)