Publish reviewed bounded JWT ready handoff and retained owner history

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade

scripts/openbao_negative_private_handoff.py

@@ -0,0 +1,266 @@
+#!/usr/bin/env python3
+"""Owner-held verified KeyCape CLI files -> inherited pipe; never normal output.
+No login, secret retrieval, network, executor launch or authenticated admission.
+Decoded claims below are shape/lifetime checks, not signature verification.
+"""
+import argparse
+import base64
+import contextlib
+import fcntl
+import json
+import math
+import os
+from pathlib import Path
+import signal
+import stat
+import threading
+import time
+
+MAX_FILE = 32768
+MAX_TOKEN = 16384
+
+class Refused(Exception):
+    pass
+
+def need(condition, code):
+    if not condition:
+        raise Refused(code)
+
+def unique(pairs):
+    out = {}
+    for key, value in pairs:
+        need(key not in out, 'duplicate_json_key')
+        out[key] = value
+    return out
+
+def decode(raw):
+    try:
+        return json.loads(raw, object_pairs_hook=unique,
+            parse_constant=lambda _: (_ for _ in ()).throw(Refused('nonfinite_json')))
+    except Refused:
+        raise
+    except Exception:
+        raise Refused('invalid_json') from None
+
+@contextlib.contextmanager
+def deadline(seconds):
+    need(type(seconds) in (int, float) and math.isfinite(seconds) and 0 < seconds <= 5,
+         'invalid_delivery_budget')
+    need(threading.current_thread() is threading.main_thread(), 'main_thread_required')
+    need(signal.getitimer(signal.ITIMER_REAL) == (0.0, 0.0), 'existing_timer_refused')
+    previous = signal.getsignal(signal.SIGALRM)
+    def expired(*_):
+        raise Refused('delivery_timeout')
+    try:
+        signal.signal(signal.SIGALRM, expired)
+        signal.setitimer(signal.ITIMER_REAL, seconds)
+        yield
+    finally:
+        signal.setitimer(signal.ITIMER_REAL, 0)
+        signal.signal(signal.SIGALRM, previous)
+
+READY_SCHEMA = 'key39-private-input-ready/v1'
+READY_KEYS = {'schema','session_nonce','proposal_id','target','executor_sha256',
+              'envelope_sha256','bao_health_identity_sha256','valid_until_epoch',
+              'receive_budget_seconds'}
+
+def finite(value):
+    try:
+        return type(value) in (int,float) and math.isfinite(value)
+    except (OverflowError,ValueError):
+        return False
+
+def endpoint(fd, writable):
+    need(type(fd) is int and fd >= 3, 'private_pipe_fd_required')
+    try:
+        identity = os.fstat(fd)
+        mode = fcntl.fcntl(fd,fcntl.F_GETFL) & os.O_ACCMODE
+        need(stat.S_ISFIFO(identity.st_mode) and mode ==
+             (os.O_WRONLY if writable else os.O_RDONLY), 'private_pipe_fd_required')
+        for standard in (0,1,2):
+            try:
+                current = os.fstat(standard)
+            except OSError:
+                continue
+            need((identity.st_dev,identity.st_ino) != (current.st_dev,current.st_ino),
+                 'stdio_pipe_alias_refused')
+        return identity.st_dev,identity.st_ino
+    except OSError:
+        raise Refused('private_pipe_fd_required') from None
+
+def wait_ready(fd, expected):
+    need(isinstance(expected,dict) and set(expected) == READY_KEYS - {'schema','receive_budget_seconds'},
+         'expected_child_pins_required')
+    need(finite(expected['valid_until_epoch']) and expected['valid_until_epoch'] > time.time(),
+         'ready_window_expired')
+    raw = b''
+    while len(raw) < 4096:
+        chunk = os.read(fd,1)
+        need(chunk, 'child_ready_missing')
+        if chunk == b'\n':
+            break
+        raw += chunk
+    else:
+        raise Refused('child_ready_oversized')
+    marker = decode(raw)
+    need(isinstance(marker,dict) and set(marker) == READY_KEYS
+         and marker.get('schema') == READY_SCHEMA, 'child_ready_invalid')
+    need(all(marker.get(key) == value for key,value in expected.items()), 'child_ready_mismatch')
+    need(finite(marker['valid_until_epoch']) and finite(marker['receive_budget_seconds'])
+         and 0 < marker['receive_budget_seconds'] <= 15
+         and marker['valid_until_epoch'] > time.time(),
+         'ready_window_expired')
+    return marker
+
+def claims(token, kind, now):
+    need(isinstance(token, str) and 0 < len(token) <= MAX_TOKEN, 'invalid_token_shape')
+    parts = token.split('.')
+    need(len(parts) == 3 and all(parts), 'invalid_token_shape')
+    try:
+        obj = decode(base64.urlsafe_b64decode(parts[1] + '=' * (-len(parts[1]) % 4)))
+        need(isinstance(obj, dict), 'invalid_token_shape')
+        exp, iat = obj.get('exp'), obj.get('iat')
+        need(type(exp) in (int,float) and type(iat) in (int,float)
+             and math.isfinite(exp) and math.isfinite(iat)
+             and iat <= now + 30 and iat < exp and exp >= now + 300, 'token_window_invalid')
+        need(obj.get('iss') == 'https://kc.coulomb.social', 'token_source_invalid')
+        if kind == 'human':
+            need(obj.get('principal_type') == 'human'
+                 and obj.get('aud') == 'netkingdom-bootstrap-console'
+                 and isinstance(obj.get('sub'), str) and obj['sub'], 'human_shape_invalid')
+            assurance = obj.get('assurance', {})
+            need(isinstance(assurance,dict) and assurance.get('level') == 'aal2'
+                 and assurance.get('mfa') is True, 'human_mfa_shape_invalid')
+        else:
+            need(obj.get('principal_type') == 'service'
+                 and obj.get('aud') == 'codex-railiance-platform'
+                 and obj.get('sub') == 'service:codex:railiance-platform'
+                 and obj.get('tenant') == 'tenant:platform'
+                 and obj.get('roles') == ['coding-agent']
+                 and obj.get('scope') == 'openbao:login', 'coding_agent_shape_invalid')
+    except Refused:
+        raise
+    except Exception:
+        raise Refused('invalid_token_shape') from None
+    return token
+
+def load(path, kind, now):
+    p = Path(path)
+    need(p.is_absolute() and p.parent.resolve() == p.parent, 'private_path_required')
+    need(not any((ancestor / '.git').is_file() or (ancestor / '.git' / 'HEAD').exists()
+                 for ancestor in p.parents), 'git_path_refused')
+    directory = os.open(p.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+    try:
+        ds = os.fstat(directory)
+        need(ds.st_uid == os.getuid() and stat.S_IMODE(ds.st_mode) == 0o700,
+             'private_directory_required')
+        fd = os.open(p.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory)
+        try:
+            original = os.fstat(fd)
+            need(stat.S_ISREG(original.st_mode) and original.st_uid == os.getuid()
+                 and stat.S_IMODE(original.st_mode) == 0o600 and original.st_nlink == 1
+                 and 0 < original.st_size <= MAX_FILE, 'private_file_required')
+            raw = b''
+            while len(raw) <= MAX_FILE:
+                chunk = os.read(fd, MAX_FILE + 1 - len(raw))
+                if not chunk:
+                    break
+                raw += chunk
+            need(len(raw) <= MAX_FILE, 'private_file_oversized')
+            after = os.fstat(fd)
+            need((original.st_size,original.st_mtime_ns,original.st_ctime_ns) ==
+                 (after.st_size,after.st_mtime_ns,after.st_ctime_ns), 'private_file_changed')
+            obj = decode(raw)
+            need(isinstance(obj,dict) and obj.get('token_type','').lower() == 'bearer',
+                 'invalid_cli_token_file')
+            token = claims(obj.get('access_token'), kind, now)
+            return token, directory, p.name, original
+        finally:
+            os.close(fd)
+    except BaseException:
+        os.close(directory)
+        raise
+
+def consume(item):
+    # Comparison is not atomic inode-conditioned unlink. Continuous exclusive
+    # same-UID directory custody is mandatory; concurrent writers are outside scope.
+    _, directory, name, original = item
+    current = os.stat(name, dir_fd=directory, follow_symlinks=False)
+    need((current.st_dev,current.st_ino,current.st_uid,current.st_nlink,current.st_size,
+          current.st_mtime_ns,current.st_ctime_ns) ==
+         (original.st_dev,original.st_ino,original.st_uid,original.st_nlink,original.st_size,
+          original.st_mtime_ns,original.st_ctime_ns), 'private_file_changed')
+    os.unlink(name, dir_fd=directory)
+
+def deliver(human_path, agent_path, fd, *, ready_fd=None, expected=None,
+            consume_owner_files=False, budget=5):
+    receipt = {'adapter_performs_native_backend_actions': False, 'native_accepted': False,
+               'private_input_delivery_confirmed': False, 'caller_path_removal_completed': False,
+               'exclusive_same_uid_custody_required': True, 'atomic_inode_delete_guaranteed': False,
+               'child_ready_matched': False,
+               'credential_files_remain_possible': True, 'credential_copy_cleanup_verified': False,
+               'private_input_delivery_unknown': False}
+    items = []
+    writing = False
+    try:
+        need(consume_owner_files is True, 'explicit_owner_file_consumption_required')
+        data_identity = endpoint(fd,True)
+        control_identity = endpoint(ready_fd,False)
+        need(data_identity != control_identity, 'data_control_pipe_alias_refused')
+        need(isinstance(expected,dict) and finite(expected.get('valid_until_epoch')),
+             'expected_child_pins_required')
+        budget = min(budget, expected['valid_until_epoch'] - time.time())
+        with deadline(budget):
+            marker = wait_ready(ready_fd,expected)
+            receipt['child_ready_matched'] = True
+            remaining_alarm = signal.getitimer(signal.ITIMER_REAL)[0]
+            remaining = min(remaining_alarm, marker['receive_budget_seconds'],
+                            expected['valid_until_epoch'] - time.time())
+            need(remaining > 0, 'ready_window_expired')
+            # Clip the existing timer; never restart or extend its original budget.
+            signal.setitimer(signal.ITIMER_REAL, remaining)
+            delivery_deadline = time.monotonic() + remaining
+            now = time.time()
+            items.append(load(human_path, 'human', now))
+            items.append(load(agent_path, 'coding-agent', now))
+            need((items[0][3].st_dev,items[0][3].st_ino) !=
+                 (items[1][3].st_dev,items[1][3].st_ino), 'distinct_files_required')
+            payload = json.dumps({'human_token':items[0][0], 'coding_agent_token':items[1][0]},
+                                 separators=(',', ':')).encode() + b'\n'
+            # Unlink inspected caller-owned inodes; Python memory is not a zeroization guarantee.
+            need(time.time() < expected['valid_until_epoch'] and
+                 time.monotonic() < delivery_deadline, 'ready_window_expired')
+            for item in items:
+                consume(item)
+            receipt.update(caller_path_removal_completed=True)
+            writing = True
+            offset = 0
+            while offset < len(payload):
+                need(time.time() < expected['valid_until_epoch'] and
+                     time.monotonic() < delivery_deadline, 'ready_window_expired')
+                written = os.write(fd, payload[offset:])
+                need(written > 0, 'private_delivery_failed')
+                offset += written
+            need(time.time() < expected['valid_until_epoch'] and
+                 time.monotonic() < delivery_deadline, 'ready_window_expired')
+            receipt['private_input_delivery_confirmed'] = True
+    except Refused as error:
+        receipt['failure_code'] = str(error)
+    except Exception:
+        receipt['failure_code'] = 'private_delivery_failed'
+    finally:
+        if writing and not receipt['private_input_delivery_confirmed']:
+            receipt['private_input_delivery_unknown'] = True
+        for item in items:
+            os.close(item[1])
+        # Launcher owns these endpoints: EOF even after partial delivery.
+        for owned in {fd,ready_fd}:
+            if type(owned) is int and owned >= 3:
+                try:
+                    os.close(owned)
+                except OSError:
+                    pass
+    return receipt
+
+# Deliberately no standalone ready-file/ready-boolean CLI. Only the reviewed
+# source launcher supplies its own captured child-control endpoint and exact pins.

scripts/openbao_negative_private_launcher.py

@@ -0,0 +1,238 @@
+#!/usr/bin/env python3
+"""Founder-held, source-pinned direct SSH launcher; no credentials on normal IO.
+Ready marker is a procedural readiness check, NEVER authenticated live authority.
+Use only after exact source/host/transport/client/custody/file allowance admission.
+"""
+import argparse
+import hashlib
+import importlib.util
+import json
+import os
+from pathlib import Path
+import re
+import secrets
+import shlex
+import subprocess
+import stat
+import time
+
+ADAPTER = Path(__file__).with_name('openbao_negative_private_handoff.py')
+spec = importlib.util.spec_from_file_location('private_handoff', ADAPTER)
+handoff = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(handoff)
+PROPOSAL = 'KEY-WP-0039-T02/attended-lifecycle-v9-ready-input'
+TARGET = 'sso/keycape'
+PREPARATION_TOTAL_SECONDS = 5
+MAX_PUBLIC_BYTES = 65536
+MAX_SOURCE_BYTES = 262144
+CONTRACT_KEYS = {'schema','proposal_id','target','remote_host','remote_wrapper_path',
+    'remote_wrapper_sha256','remote_child_path','remote_child_sha256','remote_envelope_path',
+    'local_envelope_path','remote_receipt_path','founder_decision_id',
+    'private_tmpfs_file_allowance_receipt','local_launcher_sha256','local_adapter_sha256'}
+
+def read_stable(path, limit):
+    # Always called under the shared preparation timer in the actual CLI graph.
+    # NONBLOCK + regular-file check refuses FIFO/device. Resolution refuses
+    # observed symlink ancestors; metadata compares detect observed source drift.
+    p = Path(path)
+    handoff.need(p.is_absolute() and p.resolve() == p, 'source_regular_path_required')
+    fd = os.open(p,os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+    try:
+        before = os.fstat(fd)
+        handoff.need(stat.S_ISREG(before.st_mode) and before.st_nlink == 1
+                     and not stat.S_IMODE(before.st_mode) & 0o022
+                     and 0 < before.st_size <= limit, 'source_file_invalid')
+        chunks,size = [],0
+        while size <= limit:
+            chunk = os.read(fd,min(4096,limit+1-size))
+            if not chunk:
+                break
+            size += len(chunk)
+            handoff.need(size <= limit,'source_file_oversized')
+            chunks.append(chunk)
+        after = os.fstat(fd)
+        fields = lambda v:(v.st_dev,v.st_ino,v.st_uid,v.st_mode,v.st_nlink,
+                         v.st_size,v.st_mtime_ns,v.st_ctime_ns)
+        handoff.need(fields(before) == fields(after) and size == before.st_size,
+                     'source_file_changed')
+        return b''.join(chunks)
+    finally:
+        os.close(fd)
+
+def digest(path):
+    return hashlib.sha256(read_stable(path,MAX_SOURCE_BYTES)).hexdigest()
+
+def remaining_preparation(preparation_deadline):
+    handoff.need(handoff.finite(preparation_deadline),'preparation_window_invalid')
+    remaining = preparation_deadline-time.monotonic()
+    handoff.need(0 < remaining <= PREPARATION_TOTAL_SECONDS,'preparation_window_expired')
+    return remaining
+
+def pin(value):
+    return isinstance(value,str) and re.fullmatch('[0-9a-f]{64}',value) is not None
+
+def remote_path(value):
+    handoff.need(isinstance(value,str) and len(value) <= 256
+                 and re.fullmatch('/[A-Za-z0-9_./-]+',value) is not None
+                 and '..' not in Path(value).parts and not value.endswith('/'), 'remote_path_invalid')
+    return value
+
+def read_public(path):
+    raw = read_stable(path,MAX_PUBLIC_BYTES)
+    obj = handoff.decode(raw)
+    handoff.need(isinstance(obj,dict),'source_contract_invalid')
+    return raw,obj
+
+def validate_contract(contract):
+    handoff.need(isinstance(contract,dict) and set(contract) == CONTRACT_KEYS
+                 and contract['schema'] == 'key39-private-launch-contract/v1'
+                 and contract['proposal_id'] == PROPOSAL and contract['target'] == TARGET
+                 and contract['remote_host'] == 'railiance01','source_contract_invalid')
+    for field in ('remote_wrapper_path','remote_child_path','remote_envelope_path','remote_receipt_path'):
+        remote_path(contract[field])
+    handoff.need(len({contract[key] for key in ('remote_wrapper_path','remote_child_path',
+                 'remote_envelope_path','remote_receipt_path')}) == 4,'remote_paths_must_differ')
+    for field in ('remote_wrapper_sha256','remote_child_sha256','local_launcher_sha256','local_adapter_sha256'):
+        handoff.need(pin(contract[field]),'source_pin_required')
+    handoff.need(digest(__file__) == contract['local_launcher_sha256']
+                 and digest(ADAPTER) == contract['local_adapter_sha256'],'local_source_pin_changed')
+    handoff.need(all(isinstance(contract[k],str) and 0 < len(contract[k]) <= 256
+                 for k in ('founder_decision_id','private_tmpfs_file_allowance_receipt')),
+                 'founder_receipt_required')
+    raw,envelope = read_public(contract['local_envelope_path'])
+    handoff.need(envelope.get('proposal_id') == PROPOSAL and envelope.get('target') == TARGET
+                 and envelope.get('disposition') == 'approved'
+                 and envelope.get('founder_decision_id') == contract['founder_decision_id'],
+                 'source_envelope_mismatch')
+    expiry = envelope.get('valid_until_epoch')
+    window = envelope.get('maximum_session_seconds')
+    handoff.need(handoff.finite(expiry) and type(window) is int and 420 <= window <= 900
+                 and 0 < expiry-time.time() <= window,'source_window_invalid')
+    health = envelope.get('bao_health_identity_sha256')
+    handoff.need(pin(health),'source_health_pin_required')
+    hashes = envelope.get('source_hashes',{})
+    handoff.need(isinstance(hashes,dict) and
+                 hashes.get(contract['remote_child_path']) == contract['remote_child_sha256'] and
+                 hashes.get(contract['remote_wrapper_path']) == contract['remote_wrapper_sha256'],
+                 'remote_source_pin_missing')
+    owners = envelope.get('owner_acknowledgments',{})
+    handoff.need(isinstance(owners,dict) and all(isinstance(owners.get(k),str) and owners[k]
+                 for k in ('key-cape','net-kingdom','railiance-platform')),'owner_source_receipts_required')
+    return hashlib.sha256(raw).hexdigest(),envelope
+
+def launch_and_deliver(contract,human_path,agent_path,*,consume_owner_files=False,
+                       preparation_deadline=None):
+    receipt = {'native_accepted':False,'authenticated_authority_established_by_helper':False,
+               'child_ready_matched':False,'private_input_delivery_confirmed':False,
+               'consumer_process_completed':False,'consumer_outcome_unknown':False}
+    process = None
+    owned_fds = set()
+    if preparation_deadline is None:
+        preparation_deadline = time.monotonic() + PREPARATION_TOTAL_SECONDS
+    try:
+        with handoff.deadline(remaining_preparation(preparation_deadline)):
+            envelope_hash,envelope = validate_contract(contract)
+            nonce = secrets.token_hex(32)
+            expected = {'session_nonce':nonce,'proposal_id':PROPOSAL,'target':TARGET,
+                'executor_sha256':contract['remote_child_sha256'],'envelope_sha256':envelope_hash,
+                'bao_health_identity_sha256':envelope['bao_health_identity_sha256'],
+                'valid_until_epoch':envelope['valid_until_epoch']}
+            command = ['python3',contract['remote_wrapper_path'],
+                '--child',contract['remote_child_path'],'--child-sha256',contract['remote_child_sha256'],
+                '--envelope',contract['remote_envelope_path'],'--envelope-sha256',envelope_hash,
+                '--receipt',contract['remote_receipt_path'],'--session-nonce',nonce]
+            argv = ['ssh','-T','-oBatchMode=yes','-oStrictHostKeyChecking=yes',
+                '-oForwardAgent=no','-oClearAllForwardings=yes','-oConnectTimeout=5',
+                '-oServerAliveInterval=5','-oServerAliveCountMax=1',
+                'railiance01',shlex.join(command)]
+            # Do not pass arbitrary confidential producer environment to SSH.
+            allowed = ('PATH','HOME','USER','LOGNAME','LANG','LC_ALL','SSH_AUTH_SOCK')
+            environment = {key:os.environ[key] for key in allowed if key in os.environ}
+            process = subprocess.Popen(argv,stdin=subprocess.PIPE,stdout=subprocess.PIPE,
+                stderr=subprocess.DEVNULL,env=environment,close_fds=True)
+        remaining_preparation(preparation_deadline)
+        # Distinct process-owned pipes; no arbitrary caller ready file or boolean.
+        write_fd = os.dup(process.stdin.fileno()); owned_fds.add(write_fd)
+        ready_fd = os.dup(process.stdout.fileno()); owned_fds.add(ready_fd)
+        process.stdin.close()
+        # Close parent's duplicate read endpoint; only owned ready_fd is used.
+        process.stdout.close()
+        delivery = handoff.deliver(human_path,agent_path,write_fd,ready_fd=ready_fd,
+            expected=expected,consume_owner_files=consume_owner_files,
+            budget=remaining_preparation(preparation_deadline))
+        owned_fds.clear()  # deliver owns and closes both endpoints
+        receipt.update(delivery)
+        # Private bytes may already have reached the remote child even if local
+        # write completion/EOF confirmation failed. Local SSH teardown proves no
+        # remote no-effects or cleanup outcome; never clear this unknown here.
+        if delivery['private_input_delivery_unknown']:
+            receipt['consumer_outcome_unknown'] = True
+            receipt['consumer_process_completed'] = False
+        if not delivery['private_input_delivery_confirmed']:
+            return receipt
+        # Value-free exit status only; child receipt and native acceptance remain owner duties.
+        wait_budget = max(0,min(envelope['valid_until_epoch']-time.time(),
+                               envelope['maximum_session_seconds']-180)) + 330
+        try:
+            rc = process.wait(timeout=wait_budget)
+        except subprocess.TimeoutExpired:
+            receipt.update(failure_code='consumer_completion_unconfirmed',consumer_outcome_unknown=True)
+            return receipt
+        receipt['consumer_process_completed'] = rc == 0
+        receipt['consumer_outcome_unknown'] = rc != 0
+        if rc != 0:
+            receipt['failure_code'] = 'consumer_completion_unconfirmed'
+    except handoff.Refused as error:
+        receipt['failure_code'] = str(error)
+    except Exception:
+        receipt['failure_code'] = 'private_launcher_incomplete'
+        if receipt['private_input_delivery_confirmed']:
+            receipt['consumer_outcome_unknown'] = True
+    finally:
+        for fd in owned_fds:
+            try:
+                os.close(fd)
+            except OSError:
+                pass
+        if process is not None:
+            # Local teardown never claims remote process/custody reconciliation.
+            try:
+                if process.poll() is None:
+                    process.terminate()
+                    try:
+                        process.wait(timeout=1)
+                    except subprocess.TimeoutExpired:
+                        process.kill()
+                        process.wait(timeout=1)
+                for stream in (process.stdin,process.stdout):
+                    if stream is not None:
+                        stream.close()
+            except Exception:
+                receipt['transport_cleanup_unconfirmed'] = True
+                receipt['consumer_outcome_unknown'] = True
+    return receipt
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--contract',required=True)
+    parser.add_argument('--human-file',required=True)
+    parser.add_argument('--agent-file',required=True)
+    parser.add_argument('--consume-owner-files',action='store_true')
+    args = parser.parse_args()
+    preparation_deadline = time.monotonic() + PREPARATION_TOTAL_SECONDS
+    try:
+        with handoff.deadline(remaining_preparation(preparation_deadline)):
+            _,contract = read_public(args.contract)
+        receipt = launch_and_deliver(contract,args.human_file,args.agent_file,
+            consume_owner_files=args.consume_owner_files,preparation_deadline=preparation_deadline)
+    except handoff.Refused as error:
+        receipt = {'native_accepted':False,'private_input_delivery_confirmed':False,
+                   'failure_code':str(error)}
+    except Exception:
+        receipt = {'native_accepted':False,'private_input_delivery_confirmed':False,
+                   'failure_code':'private_launcher_contract_invalid'}
+    print(json.dumps(receipt,separators=(',',':')))
+    return 0 if receipt.get('consumer_process_completed') is True else 1
+
+if __name__ == '__main__':
+    raise SystemExit(main())

scripts/test_openbao_negative_private_handoff.py

@@ -0,0 +1,228 @@
+import base64
+import importlib.util
+import json
+import os
+from pathlib import Path
+import signal
+import tempfile
+import time
+import unittest
+from unittest.mock import patch
+REAL_WRITE=os.write
+spec = importlib.util.spec_from_file_location('handoff',Path(__file__).with_name('openbao_negative_private_handoff.py'))
+s = importlib.util.module_from_spec(spec);spec.loader.exec_module(s)
+
+def jwt(kind='human', **changes):
+    obj={'iss':'https://kc.coulomb.social','iat':time.time()-1,'exp':time.time()+700,
+         'principal_type':'human','aud':'netkingdom-bootstrap-console','sub':'fixture-person',
+         'assurance':{'level':'aal2','mfa':True}}
+    if kind == 'agent':
+        obj.update(principal_type='service',aud='codex-railiance-platform',
+                   sub='service:codex:railiance-platform',tenant='tenant:platform',
+                   roles=['coding-agent'],scope='openbao:login')
+    obj.update(changes)
+    body=base64.urlsafe_b64encode(json.dumps(obj).encode()).decode().rstrip('=')
+    return 'fixture.'+body+'.fixture'
+
+class Tests(unittest.TestCase):
+    def setUp(self):
+        self.tmp=tempfile.TemporaryDirectory();self.dir=Path(self.tmp.name)
+        self.h=self.dir/'human.json';self.a=self.dir/'agent.json'
+        self.write(self.h,jwt());self.write(self.a,jwt('agent'))
+    def tearDown(self):
+        self.tmp.cleanup()
+    def write(self,p,token):
+        p.write_text(json.dumps({'access_token':token,'token_type':'Bearer','expires_in':700}));p.chmod(0o600)
+    def run_delivery(self,**kw):
+        r,w=os.pipe();cr,cw=os.pipe()
+        expected=self.expected()
+        REAL_WRITE(cw,json.dumps(dict(expected,schema=s.READY_SCHEMA,receive_budget_seconds=10)).encode()+b'\n')
+        os.close(cw)
+        receipt=s.deliver(kw.pop("human_path",str(self.h)),str(self.a),w,ready_fd=cr,expected=expected,consume_owner_files=True,**kw)
+        raw=os.read(r,65536);os.close(r)
+        return receipt,raw
+    def expected(self):
+        return {'session_nonce':'1'*64,'proposal_id':'KEY-WP-0039-T02/attended-lifecycle-v9-ready-input',
+                'target':'sso/keycape','executor_sha256':'2'*64,'envelope_sha256':'3'*64,
+                'bao_health_identity_sha256':'4'*64,'valid_until_epoch':time.time()+600}
+    def test_success_only_private_pipe_and_consumes_owned_files(self):
+        receipt,raw=self.run_delivery()
+        obj=json.loads(raw)
+        self.assertEqual(set(obj),{'human_token','coding_agent_token'})
+        self.assertTrue(receipt['private_input_delivery_confirmed'])
+        self.assertFalse(receipt['native_accepted'])
+        self.assertNotIn('fixture',json.dumps(receipt))
+        self.assertFalse(self.h.exists());self.assertFalse(self.a.exists())
+    def test_stdout_stderr_and_regular_fd_refused_before_file_consumption(self):
+        for fd in (1,2):
+            receipt=s.deliver(str(self.h),str(self.a),fd,consume_owner_files=True)
+            self.assertEqual(receipt['failure_code'],'private_pipe_fd_required')
+        with tempfile.TemporaryFile() as f:
+            dup=os.dup(f.fileno())
+            receipt=s.deliver(str(self.h),str(self.a),dup,consume_owner_files=True)
+            self.assertEqual(receipt['failure_code'],'private_pipe_fd_required')
+            self.assertEqual(f.tell(),0)
+        self.assertTrue(self.h.exists());self.assertTrue(self.a.exists())
+    def test_read_only_pipe_refused_before_consuming_files(self):
+        r,w=os.pipe()
+        receipt=s.deliver(str(self.h),str(self.a),r,consume_owner_files=True)
+        os.close(w)
+        self.assertEqual(receipt['failure_code'],'private_pipe_fd_required')
+        self.assertTrue(self.h.exists());self.assertTrue(self.a.exists())
+    def test_git_worktree_refused_before_delivery(self):
+        marker=self.dir/'.git';marker.write_text('gitdir: fixture')
+        receipt,raw=self.run_delivery()
+        self.assertEqual(receipt['failure_code'],'git_path_refused');self.assertFalse(raw)
+    def test_symlinked_parent_refused(self):
+        alias=self.dir/'alias';alias.symlink_to(self.dir,target_is_directory=True)
+        receipt,raw=self.run_delivery(human_path=str(alias/'human.json'))
+        self.assertEqual(receipt['failure_code'],'private_path_required')
+        self.assertFalse(raw);self.assertTrue(self.h.exists())
+
+    def test_explicit_consumption_required(self):
+        r,w=os.pipe();receipt=s.deliver(str(self.h),str(self.a),w)
+        self.assertEqual(receipt['failure_code'],'explicit_owner_file_consumption_required')
+        self.assertEqual(os.read(r,10),b'');os.close(r);self.assertTrue(self.h.exists())
+    def test_symlink_refused_no_delivery(self):
+        target=self.dir/'target';self.h.rename(target);self.h.symlink_to(target)
+        receipt,raw=self.run_delivery();self.assertFalse(raw)
+        self.assertFalse(receipt['private_input_delivery_confirmed']);self.assertTrue(target.exists())
+    def test_hardlinked_file_refused(self):
+        os.link(self.h,self.dir/'extra');receipt,raw=self.run_delivery()
+        self.assertEqual(receipt['failure_code'],'private_file_required');self.assertFalse(raw)
+    def test_public_mode_refused(self):
+        self.a.chmod(0o644);receipt,raw=self.run_delivery()
+        self.assertEqual(receipt['failure_code'],'private_file_required');self.assertFalse(raw)
+    def test_public_directory_refused(self):
+        self.dir.chmod(0o755);receipt,raw=self.run_delivery()
+        self.assertEqual(receipt['failure_code'],'private_directory_required');self.assertFalse(raw)
+    def test_oversize_file_refused_before_delivery(self):
+        self.a.write_bytes(b'x'*(s.MAX_FILE+1));receipt,raw=self.run_delivery()
+        self.assertEqual(receipt['failure_code'],'private_file_required');self.assertFalse(raw)
+    def test_wrong_human_audience_or_missing_mfa_refused(self):
+        for changes in ({'aud':'openbao-admin'},{'assurance':{'level':'aal1','mfa':False}}):
+            self.write(self.h,jwt(**changes));receipt,raw=self.run_delivery()
+            self.assertFalse(raw);self.assertFalse(receipt['private_input_delivery_confirmed'])
+    def test_approle_string_or_wrong_coding_claim_refused(self):
+        for token in ('hvs.not-a-jwt',jwt('agent',tenant='tenant:coulomb'),jwt('agent',roles=['secrets-engine'])):
+            self.write(self.a,token);receipt,raw=self.run_delivery()
+            self.assertFalse(raw);self.assertFalse(receipt['private_input_delivery_confirmed'])
+    def test_expired_bool_or_infinite_timestamp_refused(self):
+        for expiry in (time.time()-1,True,float('inf')):
+            self.write(self.a,jwt('agent',exp=expiry));receipt,raw=self.run_delivery()
+            self.assertFalse(raw);self.assertFalse(receipt['private_input_delivery_confirmed'])
+    def test_duplicate_json_refused(self):
+        self.a.write_text('{"token_type":"Bearer","access_token":"a","access_token":"b"}')
+        receipt,raw=self.run_delivery();self.assertEqual(receipt['failure_code'],'duplicate_json_key');self.assertFalse(raw)
+    def test_file_replacement_before_consumption_not_unlinked(self):
+        original=s.consume
+        def replace(item):
+            self.h.unlink();self.write(self.h,jwt())
+            return original(item)
+        with patch.object(s,'consume',side_effect=replace):receipt,raw=self.run_delivery()
+        self.assertEqual(receipt['failure_code'],'private_file_changed');self.assertFalse(raw);self.assertTrue(self.h.exists())
+    def test_blocked_partial_write_is_bounded_unknown_and_timer_restored(self):
+        previous=signal.getsignal(signal.SIGALRM);started=time.monotonic()
+        with patch.object(s.os,'write',side_effect=lambda *args:time.sleep(0.1)):
+            receipt,raw=self.run_delivery(budget=0.01)
+        self.assertLess(time.monotonic()-started,0.06)
+        self.assertEqual(receipt['failure_code'],'delivery_timeout')
+        self.assertTrue(receipt['private_input_delivery_unknown']);self.assertFalse(raw)
+        self.assertFalse(self.h.exists());self.assertFalse(self.a.exists())
+        self.assertEqual(signal.getitimer(signal.ITIMER_REAL),(0.0,0.0));self.assertIs(signal.getsignal(signal.SIGALRM),previous)
+    def test_broken_pipe_fixed_error_no_raw_exception(self):
+        with patch.object(s.os,'write',side_effect=RuntimeError('protected raw value')):
+            receipt,raw=self.run_delivery()
+        self.assertEqual(receipt['failure_code'],'private_delivery_failed')
+        self.assertTrue(receipt['private_input_delivery_unknown']);self.assertNotIn('protected',json.dumps(receipt));self.assertFalse(raw)
+
+if __name__ == '__main__':unittest.main()
+
+class ReadyTests(unittest.TestCase):
+    setUp=Tests.setUp
+    tearDown=Tests.tearDown
+    write=Tests.write
+    expected=Tests.expected
+    run_delivery=Tests.run_delivery
+    # Concrete ready-channel counterexamples; base safety cases run only once.
+    def with_marker(self, marker, budget=0.05):
+        r,w=os.pipe();cr,cw=os.pipe();expected=self.expected()
+        raw=marker(expected) if callable(marker) else marker
+        REAL_WRITE(cw,raw);os.close(cw)
+        with patch.object(s,'load',wraps=s.load) as load:
+            receipt=s.deliver(str(self.h),str(self.a),w,ready_fd=cr,expected=expected,
+                              consume_owner_files=True,budget=budget)
+            calls=load.call_count
+        data=os.read(r,65536);os.close(r)
+        return receipt,data,calls
+    def test_no_child_ready_or_mismatch_means_zero_file_load_and_private_bytes(self):
+        for marker in (b'', b'not-json\n', lambda e:json.dumps(dict(e,schema=s.READY_SCHEMA,
+                receive_budget_seconds=10,session_nonce='0'*64)).encode()+b'\n',
+                lambda e:json.dumps(dict(e,schema=s.READY_SCHEMA,receive_budget_seconds=True)).encode()+b'\n'):
+            receipt,data,calls=self.with_marker(marker)
+            self.assertFalse(receipt['private_input_delivery_confirmed']);self.assertFalse(data)
+            self.assertEqual(calls,0);self.assertTrue(self.h.exists());self.assertTrue(self.a.exists())
+    def test_no_ready_eof_hard_timeout_zero_file_load_or_consumption(self):
+        r,w=os.pipe();cr,cw=os.pipe();started=time.monotonic()
+        try:
+            with patch.object(s,'load') as load:
+                receipt=s.deliver(str(self.h),str(self.a),w,ready_fd=cr,expected=self.expected(),
+                                  consume_owner_files=True,budget=0.01)
+                load.assert_not_called()
+            self.assertLess(time.monotonic()-started,0.06)
+            self.assertEqual(receipt['failure_code'],'delivery_timeout');self.assertFalse(os.read(r,10))
+            self.assertTrue(self.h.exists());self.assertTrue(self.a.exists())
+        finally:os.close(cw);os.close(r)
+    def test_stdout_and_stderr_pipe_duplicates_refused(self):
+        for standard in (1,2):
+            r,w=os.pipe();backup=os.dup(standard)
+            try:
+                os.dup2(w,standard);duplicate=os.dup(standard)
+                receipt=s.deliver(str(self.h),str(self.a),duplicate,consume_owner_files=True)
+            finally:os.dup2(backup,standard);os.close(backup);os.close(w)
+            self.assertEqual(receipt['failure_code'],'stdio_pipe_alias_refused')
+            self.assertEqual(os.read(r,10),b'');os.close(r)
+            self.assertTrue(self.h.exists());self.assertTrue(self.a.exists())
+    def test_data_and_control_same_pipe_refused(self):
+        r,w=os.pipe();readback=os.dup(r)
+        receipt=s.deliver(str(self.h),str(self.a),w,ready_fd=r,expected=self.expected(),consume_owner_files=True)
+        self.assertEqual(receipt['failure_code'],'data_control_pipe_alias_refused')
+        self.assertEqual(os.read(readback,10),b'');os.close(readback)
+    def test_oversize_ready_refused_before_load(self):
+        receipt,data,calls=self.with_marker(b'x'*4096+b'\n')
+        self.assertEqual(receipt['failure_code'],'child_ready_oversized');self.assertFalse(data);self.assertEqual(calls,0)
+    def test_received_short_budget_clips_timer_before_token_load(self):
+        marker=lambda e:json.dumps(dict(e,schema=s.READY_SCHEMA,receive_budget_seconds=0.01)).encode()+b'\n'
+        original=s.load
+        def delayed(*args):time.sleep(0.1);return original(*args)
+        started=time.monotonic()
+        with patch.object(s,'load',side_effect=delayed):receipt,data,calls=self.with_marker(marker)
+        self.assertLess(time.monotonic()-started,0.06);self.assertFalse(data)
+        self.assertEqual(receipt['failure_code'],'delivery_timeout');self.assertTrue(self.h.exists())
+    def test_real_partial_write_stays_unknown_and_conservative_cleanup(self):
+        import fcntl
+        self.write(self.h,jwt(padding='h'*7000));self.write(self.a,jwt('agent',padding='a'*7000))
+        r,w=os.pipe();fcntl.fcntl(w,fcntl.F_SETPIPE_SZ,4096)
+        cr,cw=os.pipe();e=self.expected();REAL_WRITE(cw,json.dumps(dict(e,schema=s.READY_SCHEMA,
+            receive_budget_seconds=10)).encode()+b'\n');os.close(cw)
+        receipt=s.deliver(str(self.h),str(self.a),w,ready_fd=cr,expected=e,consume_owner_files=True,budget=0.015)
+        data=os.read(r,65536);os.close(r)
+        self.assertEqual(len(data),4096);self.assertTrue(receipt['private_input_delivery_unknown'])
+        self.assertFalse(receipt['private_input_delivery_confirmed'])
+        self.assertTrue(receipt['credential_files_remain_possible'])
+        self.assertFalse(receipt['credential_copy_cleanup_verified'])
+    def test_stat_unlink_race_retained_as_exclusive_custody_violation_not_atomic_claim(self):
+        original=s.os.stat;replaced=[]
+        def race(path,*args,**kwargs):
+            current=original(path,*args,**kwargs)
+            if path=='human.json' and kwargs.get('dir_fd') is not None and not replaced:
+                self.h.unlink();self.write(self.h,jwt(sub='replacement-fixture'))
+                replaced.append(True)
+            return current
+        with patch.object(s.os,'stat',side_effect=race):receipt,data=self.run_delivery()
+        self.assertTrue(replaced);self.assertFalse(self.h.exists())
+        self.assertTrue(receipt['caller_path_removal_completed'])
+        self.assertFalse(receipt['atomic_inode_delete_guaranteed'])
+        self.assertTrue(receipt['exclusive_same_uid_custody_required'])
+        self.assertTrue(receipt['credential_files_remain_possible'])
+        self.assertFalse(receipt['credential_copy_cleanup_verified'])

scripts/test_openbao_negative_private_launcher.py

@@ -0,0 +1,145 @@
+import importlib.util
+import json
+import os
+from pathlib import Path
+import shlex
+import tempfile
+import time
+import unittest
+from unittest.mock import patch
+spec=importlib.util.spec_from_file_location('launcher',Path(__file__).with_name('openbao_negative_private_launcher.py'))
+l=importlib.util.module_from_spec(spec);spec.loader.exec_module(l)
+spec=importlib.util.spec_from_file_location('fixtures',Path(__file__).with_name('test_openbao_negative_private_handoff.py'))
+f=importlib.util.module_from_spec(spec);spec.loader.exec_module(f)
+
+class FakeProcess:
+    def __init__(self,argv,contract,envelope_hash,envelope,marker='valid'):
+        self.argv=argv;self.rc=None;self.terminated=False;self.waits=[]
+        self.read_data,w=os.pipe();self.stdin=os.fdopen(w,'wb',buffering=0)
+        r,cw=os.pipe();self.stdout=os.fdopen(r,'rb',buffering=0)
+        args=shlex.split(argv[-1]);nonce=args[args.index('--session-nonce')+1]
+        ready={'schema':'key39-private-input-ready/v1','session_nonce':nonce,
+               'proposal_id':l.PROPOSAL,'target':l.TARGET,'executor_sha256':contract['remote_child_sha256'],
+               'envelope_sha256':envelope_hash,'bao_health_identity_sha256':envelope['bao_health_identity_sha256'],
+               'valid_until_epoch':envelope['valid_until_epoch'],'receive_budget_seconds':15}
+        if marker=='wrong':ready['session_nonce']='0'*64
+        if marker!='missing':os.write(cw,json.dumps(ready).encode()+b'\n')
+        os.close(cw)
+    def wait(self,timeout):
+        self.waits.append(timeout);self.rc=-15 if self.terminated else 0;return self.rc
+    def poll(self):return self.rc
+    def terminate(self):self.terminated=True
+    def kill(self):self.terminated=True
+
+class Tests(unittest.TestCase):
+    def setUp(self):
+        self.tmp=tempfile.TemporaryDirectory();self.dir=Path(self.tmp.name)
+        self.h=self.dir/'human.json';self.a=self.dir/'agent.json'
+        for p,token in ((self.h,f.jwt()),(self.a,f.jwt('agent'))):
+            p.write_text(json.dumps({'access_token':token,'token_type':'Bearer'}));p.chmod(0o600)
+        self.envelope={'proposal_id':l.PROPOSAL,'target':l.TARGET,'disposition':'approved',
+            'founder_decision_id':'fixture-founder','valid_until_epoch':time.time()+600,
+            'maximum_session_seconds':900,'bao_health_identity_sha256':'4'*64,
+            'source_hashes':{'/run/key39/child.py':'2'*64,'/run/key39/wrapper.py':'3'*64},
+            'owner_acknowledgments':{k:'fixture-source' for k in ('key-cape','net-kingdom','railiance-platform')}}
+        self.e=self.dir/'envelope.json';self.e.write_text(json.dumps(self.envelope))
+        self.contract={'schema':'key39-private-launch-contract/v1','proposal_id':l.PROPOSAL,
+            'target':l.TARGET,'remote_host':'railiance01','remote_wrapper_path':'/run/key39/wrapper.py',
+            'remote_wrapper_sha256':'3'*64,'remote_child_path':'/run/key39/child.py',
+            'remote_child_sha256':'2'*64,'remote_envelope_path':'/run/key39/approved.json',
+            'local_envelope_path':str(self.e),'remote_receipt_path':'/run/key39/receipt.json',
+            'founder_decision_id':'fixture-founder','private_tmpfs_file_allowance_receipt':'fixture-file-allowance',
+            'local_launcher_sha256':l.digest(l.__file__),'local_adapter_sha256':l.digest(l.ADAPTER)}
+        self.processes=[]
+    def tearDown(self):
+        for p in self.processes:
+            os.close(p.read_data)
+        self.tmp.cleanup()
+    def run_launcher(self,marker='valid'):
+        def popen(argv,**kwargs):
+            _,envelope=l.read_public(self.e)
+            self.assertEqual(kwargs['stdin'],l.subprocess.PIPE)
+            self.assertEqual(kwargs['stdout'],l.subprocess.PIPE)
+            self.assertEqual(kwargs['stderr'],l.subprocess.DEVNULL)
+            self.assertNotIn('KEYCAPE_CODEX_RAILIANCE_PLATFORM_CLIENT_SECRET',kwargs['env'])
+            self.assertIn('-oStrictHostKeyChecking=yes',argv);self.assertIn('-T',argv)
+            p=FakeProcess(argv,self.contract,l.digest(self.e),envelope,marker);self.processes.append(p);return p
+        with patch.object(l.subprocess,'Popen',side_effect=popen) as spawn, \
+             patch.dict(os.environ,{'KEYCAPE_CODEX_RAILIANCE_PLATFORM_CLIENT_SECRET':'PRIVATE-FIXTURE'}):
+            receipt=l.launch_and_deliver(self.contract,str(self.h),str(self.a),consume_owner_files=True)
+        return receipt,spawn.call_count
+    def test_complete_pinned_child_marker_then_private_pipe_delivery_no_actual_ssh(self):
+        receipt,count=self.run_launcher();self.assertEqual(count,1)
+        self.assertTrue(receipt['child_ready_matched']);self.assertTrue(receipt['consumer_process_completed'])
+        self.assertFalse(receipt['native_accepted']);self.assertFalse(receipt['authenticated_authority_established_by_helper'])
+        raw=os.read(self.processes[0].read_data,65536);self.assertEqual(set(json.loads(raw)),{'human_token','coding_agent_token'})
+        self.assertNotIn('fixture-person',json.dumps(receipt));self.assertFalse(self.h.exists());self.assertFalse(self.a.exists())
+    def test_missing_or_wrong_marker_retains_files_and_writes_zero_tokens(self):
+        for marker in ('missing','wrong'):
+            receipt,count=self.run_launcher(marker)
+            self.assertEqual(count,1);self.assertFalse(receipt['child_ready_matched'])
+            self.assertEqual(os.read(self.processes[-1].read_data,65536),b'')
+            self.assertTrue(self.h.exists());self.assertTrue(self.a.exists())
+    def test_changed_local_source_pin_refuses_before_transport_and_file_read(self):
+        self.contract['local_adapter_sha256']='0'*64
+        receipt,count=self.run_launcher();self.assertEqual(count,0)
+        self.assertEqual(receipt['failure_code'],'local_source_pin_changed');self.assertTrue(self.h.exists())
+    def test_wrong_host_or_remote_command_path_refuses_before_transport(self):
+        for field,value in (('remote_host','other-node'),('remote_wrapper_path','/tmp/x;cat'),
+                            ('remote_child_path','/run/key39/../child.py')):
+            old=self.contract[field];self.contract[field]=value
+            receipt,count=self.run_launcher();self.contract[field]=old
+            self.assertEqual(count,0);self.assertFalse(receipt['private_input_delivery_confirmed']);self.assertTrue(self.h.exists())
+    def test_expired_or_wrong_founder_source_contract_no_spawn_no_file_access(self):
+        for key,value in (('valid_until_epoch',time.time()-1),('founder_decision_id','other')):
+            current=json.loads(self.e.read_text());current[key]=value;self.e.write_text(json.dumps(current))
+            receipt,count=self.run_launcher();self.assertEqual(count,0);self.assertTrue(self.h.exists())
+            self.e.write_text(json.dumps(self.envelope))
+
+    def test_full_pipe_payload_then_confirmation_timeout_keeps_remote_unknown(self):
+        original_deliver=l.handoff.deliver
+        original_write=os.write
+        def interrupted_delivery(*args,**kwargs):
+            kwargs['budget']=0.01
+            def write_then_stall(fd,payload):
+                written=original_write(fd,payload)
+                time.sleep(0.1)
+                return written
+            with patch.object(l.handoff.os,'write',side_effect=write_then_stall):
+                return original_deliver(*args,**kwargs)
+        with patch.object(l.handoff,'deliver',side_effect=interrupted_delivery):
+            receipt,count=self.run_launcher()
+        self.assertEqual(count,1)
+        data=os.read(self.processes[0].read_data,65536)
+        self.assertEqual(set(json.loads(data)),{'human_token','coding_agent_token'})
+        self.assertTrue(receipt['private_input_delivery_unknown'])
+        self.assertFalse(receipt['private_input_delivery_confirmed'])
+        self.assertTrue(receipt['consumer_outcome_unknown'])
+        self.assertFalse(receipt['consumer_process_completed'])
+        self.assertTrue(self.processes[0].terminated)
+        self.assertFalse(self.h.exists());self.assertFalse(self.a.exists())
+        self.assertFalse(receipt['native_accepted'])
+    def test_real_partial_pipe_write_unknown_is_sticky_after_local_terminate(self):
+        import fcntl
+        for p,token in ((self.h,f.jwt(padding='h'*7000)),(self.a,f.jwt('agent',padding='a'*7000))):
+            p.write_text(json.dumps({'access_token':token,'token_type':'Bearer'}));p.chmod(0o600)
+        original_process=FakeProcess
+        original_deliver=l.handoff.deliver
+        def small_pipe(*args,**kwargs):
+            proc=original_process(*args,**kwargs)
+            fcntl.fcntl(proc.stdin.fileno(),fcntl.F_SETPIPE_SZ,4096)
+            return proc
+        def bounded(*args,**kwargs):
+            kwargs['budget']=0.015
+            return original_deliver(*args,**kwargs)
+        with patch.dict(globals(),{'FakeProcess':small_pipe}),patch.object(l.handoff,'deliver',side_effect=bounded):
+            receipt,count=self.run_launcher()
+        data=os.read(self.processes[0].read_data,65536)
+        self.assertEqual(len(data),4096)
+        self.assertTrue(receipt['private_input_delivery_unknown'])
+        self.assertTrue(receipt['consumer_outcome_unknown'])
+        self.assertFalse(receipt['consumer_process_completed'])
+        self.assertTrue(self.processes[0].terminated)
+        self.assertFalse(receipt['native_accepted'])
+
+if __name__=='__main__':unittest.main()

scripts/test_openbao_negative_private_preparation.py

@@ -0,0 +1,91 @@
+import importlib.util
+import io
+import json
+import os
+from pathlib import Path
+import signal
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from unittest.mock import patch
+spec=importlib.util.spec_from_file_location('prep_launcher',Path(__file__).with_name('openbao_negative_private_launcher.py'))
+l=importlib.util.module_from_spec(spec);spec.loader.exec_module(l)
+
+class Tests(unittest.TestCase):
+    def actual_cli(self,path):
+        started=time.monotonic()
+        result=subprocess.run([sys.executable,l.__file__,'--contract',str(path),
+            '--human-file','/NEVER-READ-HUMAN','--agent-file','/NEVER-READ-AGENT'],
+            capture_output=True,timeout=0.4)
+        self.assertEqual(result.returncode,1);self.assertEqual(result.stderr,b'')
+        self.assertLess(time.monotonic()-started,0.25)
+        value=json.loads(result.stdout);self.assertFalse(value['native_accepted'])
+        self.assertFalse(value['private_input_delivery_confirmed'])
+        return value
+    def test_actual_main_fifo_refused_without_waiting_for_writer(self):
+        with tempfile.TemporaryDirectory() as folder:
+            path=Path(folder)/'contract.fifo';os.mkfifo(path,0o600)
+            self.assertEqual(self.actual_cli(path)['failure_code'],'source_file_invalid')
+    def test_actual_main_symlink_contract_refused(self):
+        with tempfile.TemporaryDirectory() as folder:
+            original=Path(folder)/'contract.json';original.write_text('{}')
+            path=Path(folder)/'alias';path.symlink_to(original)
+            self.assertEqual(self.actual_cli(path)['failure_code'],'source_regular_path_required')
+    def test_actual_main_oversize_contract_refused(self):
+        with tempfile.TemporaryDirectory() as folder:
+            path=Path(folder)/'contract.json';path.write_bytes(b'x'*(l.MAX_PUBLIC_BYTES+1))
+            self.assertEqual(self.actual_cli(path)['failure_code'],'source_file_invalid')
+    def main_with(self,path,**patches):
+        out=io.StringIO()
+        argv=['launcher','--contract',str(path),'--human-file','/NEVER-READ-HUMAN',
+              '--agent-file','/NEVER-READ-AGENT']
+        with patch.object(sys,'argv',argv),patch.object(sys,'stdout',out), \
+             patch.object(l.subprocess,'Popen') as spawn,patch.object(l.handoff,'load') as load:
+            rc=l.main();spawn.assert_not_called();load.assert_not_called()
+        self.assertEqual(rc,1)
+        return json.loads(out.getvalue())
+    def test_actual_main_slow_regular_read_has_hard_total_bound_and_signal_restore(self):
+        with tempfile.TemporaryDirectory() as folder:
+            path=Path(folder)/'contract.json';path.write_text('{}')
+            original=l.os.read
+            def slow(fd,size):time.sleep(0.1);return original(fd,size)
+            previous=signal.getsignal(signal.SIGALRM);started=time.monotonic()
+            with patch.object(l,'PREPARATION_TOTAL_SECONDS',0.01),patch.object(l.os,'read',side_effect=slow):
+                receipt=self.main_with(path)
+            self.assertEqual(receipt['failure_code'],'delivery_timeout')
+            self.assertLess(time.monotonic()-started,0.06)
+            self.assertEqual(signal.getitimer(signal.ITIMER_REAL),(0.0,0.0));self.assertIs(signal.getsignal(signal.SIGALRM),previous)
+    def test_actual_main_launch_does_not_renew_budget_spent_acquiring_contract(self):
+        with tempfile.TemporaryDirectory() as folder:
+            path=Path(folder)/'contract.json';path.write_text('{}')
+            original=l.read_public
+            def acquire(path):time.sleep(0.02);return original(path)
+            def slow_validate(contract):time.sleep(0.02);return '0'*64,{}
+            started=time.monotonic()
+            with patch.object(l,'PREPARATION_TOTAL_SECONDS',0.03), \
+                 patch.object(l,'read_public',side_effect=acquire), \
+                 patch.object(l,'validate_contract',side_effect=slow_validate):
+                receipt=self.main_with(path)
+            self.assertEqual(receipt['failure_code'],'delivery_timeout');self.assertLess(time.monotonic()-started,0.065)
+    def test_observed_source_change_refused_and_reads_are_bounded(self):
+        with tempfile.TemporaryDirectory() as folder:
+            path=Path(folder)/'contract.json';path.write_text('{}')
+            original=l.os.read;changed=[];sizes=[]
+            def mutate(fd,size):
+                sizes.append(size)
+                if not changed:path.write_text('{"x":1}');changed.append(True)
+                return original(fd,size)
+            with patch.object(l.os,'read',side_effect=mutate):
+                with self.assertRaisesRegex(l.handoff.Refused,'source_file_changed'):l.read_public(path)
+            self.assertTrue(all(0<size<=4096 for size in sizes))
+    def test_source_digest_fifo_and_symlink_refuse(self):
+        with tempfile.TemporaryDirectory() as folder:
+            path=Path(folder)/'source.fifo';os.mkfifo(path,0o600)
+            with self.assertRaisesRegex(l.handoff.Refused,'source_file_invalid'):l.digest(path)
+            regular=Path(folder)/'source.py';regular.write_text('fixture')
+            alias=Path(folder)/'alias';alias.symlink_to(regular)
+            with self.assertRaisesRegex(l.handoff.Refused,'source_regular_path_required'):l.digest(alias)
+
+if __name__=='__main__':unittest.main()