12 scheduled · {sum(row.get("valid") is True for row in group)}/12 valid · {sum(row["evaluation"].get("first_admission") is not None for row in group)} first admissions · {false_initial} false initial admissions · {complete}/12 with complete five-second follow-up · {native_pass} native grip holds · {stationary_pass} grip plus stationary-tool pauses · {false_confirm} false confirmations · {abstain} conservative abstentions
')
+ sections.append('DVIDIA · PiPER retention v0.2
Checking grip retention without changing the squeeze.
+Three authored hold strategies across twelve fresh synthetic scenes. The primary question is whether an admitted grasp remains held continuously for five simulated seconds without a false confirmation. A second reported outcome also requires the TCP to stay within the same five-second pose bounds for every variant.
+Simulation-only outcomes. No learned policy, transport, physical sensor validation or hardware qualification.
+{esc(report.get('scope', ''))}
{''.join(sections)}
+How to read the outcomes
“Native five-second grip hold” is the simulator outcome after first admission. “Grip + stationary tool” additionally requires the TCP to remain within 4 mm and 0.05 rad of its admission pose throughout the five-second window. “Controller retained” is the controller's corresponding qualification claim. A false confirmation is counted only when complete follow-up exists, the native grip hold fails, and the controller claims retention. Conservative abstention is shown separately when the native hold succeeds and the controller does not claim retention. Initial admission against the native prefix is a separate event. Incomplete follow-up is not counted as either mismatch.
+Step p95 is unpaced full control-step compute wall time (including dense telemetry); it is not sensor latency or a real-time guarantee. Runs use privileged simulator state and do not establish physical retention.
+'''
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument('report', type=Path)
+ parser.add_argument('--output', type=Path)
+ args = parser.parse_args()
+ report, rows = load(args.report)
+ output = args.output or args.report.with_name('report.html')
+ if output.exists() or output.is_symlink():
+ raise ValueError('HTML output path must be new; existing files are not overwritten.')
+ output.write_text(render(report, rows), encoding='utf-8')
+ print(output)
+
+
+if __name__ == '__main__':
+ main()
diff --git a/benchmarks/piper-retention-v02/test_independent_audit.py b/benchmarks/piper-retention-v02/test_independent_audit.py
new file mode 100644
index 0000000..7881dbb
--- /dev/null
+++ b/benchmarks/piper-retention-v02/test_independent_audit.py
@@ -0,0 +1,40 @@
+import importlib.util
+from pathlib import Path
+import subprocess
+import sys
+import unittest
+import numpy as np
+
+PATH=Path(__file__).with_name('independent_audit.py')
+spec=importlib.util.spec_from_file_location('retention_independent',PATH)
+audit=importlib.util.module_from_spec(spec);spec.loader.exec_module(audit)
+
+class IndependentAuditTests(unittest.TestCase):
+ def fixture(self):
+ n=5200
+ arrays={name:np.zeros((n,) if width==1 else(n,width),dtype=np.float64) for name,width in audit.FIELDS.items()}
+ arrays['t_seconds']=np.arange(1,n+1)*.001
+ arrays['object_position_m'][:]=[.3,0,.08]
+ arrays['object_position_m'][20:,2]+=.006
+ arrays['object_minus_tcp_local_m'][:]=[0,0,-.1]
+ arrays['grasp_contact_counts'][:]=1;arrays['pad_normal_forces_n'][:]=.02
+ return arrays,{'start_time_s':.02,'object_origin_m':[.3,0,.08],'local_offset_origin_m':[0,0,-.1]}
+ def test_native_100ms_and_hidden_gap(self):
+ self.assertTrue(audit.self_tests()['passed'])
+ def test_force_ceiling_in_admission_window(self):
+ a,p=self.fixture();a['pad_normal_forces_n'][100,0]=10.001
+ condition,streak,qualified,*_=audit.native_predicate(a,p)
+ self.assertFalse(condition[100]);self.assertEqual(streak[100],0);self.assertFalse(qualified[119])
+ a['pad_normal_forces_n'][100,0]=10.
+ self.assertTrue(audit.native_predicate(a,p)[2][119])
+ def test_exact_five_second_native_interval_and_gap(self):
+ a,p=self.fixture();condition,*_=audit.native_predicate(a,p)
+ start=audit.row_at(.12,a['t_seconds']);end=audit.row_at(5.12,a['t_seconds'])
+ self.assertEqual(end-start,5000);self.assertTrue(condition[start:end+1].all())
+ a['grasp_contact_counts'][end-1,1]=0
+ self.assertFalse(audit.native_predicate(a,p)[0][start:end+1].all())
+ def test_optimized_python_refused_before_import(self):
+ result=subprocess.run([sys.executable,'-O',str(PATH),'--help'],capture_output=True,text=True)
+ self.assertNotEqual(result.returncode,0);self.assertIn('refuses optimized Python',result.stderr)
+
+if __name__=='__main__':unittest.main()
diff --git a/benchmarks/piper-retention-v02/test_package_evidence.py b/benchmarks/piper-retention-v02/test_package_evidence.py
new file mode 100644
index 0000000..11aa697
--- /dev/null
+++ b/benchmarks/piper-retention-v02/test_package_evidence.py
@@ -0,0 +1,177 @@
+from hashlib import sha256
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+import zipfile
+
+import package_evidence as package
+
+
+def write_json(path, value):
+ path.write_bytes(package.encoded(value))
+
+
+class PackageEvidenceTests(unittest.TestCase):
+ def setUp(self):
+ self.temporary = tempfile.TemporaryDirectory()
+ self.addCleanup(self.temporary.cleanup)
+ self.root = Path(self.temporary.name)
+ self.study = self.root / 'piper-retention-v02-run'
+ self.audit = self.root / 'piper-retention-v02-audit'
+ self.study.mkdir(); self.audit.mkdir()
+ rows = [{'id': f'{scene}-{variant}', 'scene_id': f'scene-{scene}', 'variant': variant}
+ for scene in range(12) for variant in sorted(package.VARIANTS)]
+ write_json(self.study / 'report.json', {'schema': package.SCHEMA, 'planned_attempts': 36,
+ 'attempted': 36, 'rows': rows, 'source_sha256': {},
+ 'aggregates': {v: {k: 0 for k in ('admissions', 'false_initial_admissions',
+ 'native_5s_pause_success', 'grip_retention_and_stationarity', 'final_both_qualified')}
+ for v in package.VARIANTS}})
+ write_json(self.study / 'protocol.json', {'schema': 'synthetic-protocol', 'source_sha256': {}})
+ write_json(self.study / 'inspection.json', {
+ 'schema': 'dvidia.piper-retention-inspection/v1', 'attempts': 36,
+ 'rows': [{'id': row['id'], 'evaluation_recomputed': True} for row in rows]})
+ (self.study / 'attempt-0.bin').write_bytes(b'synthetic attempt evidence')
+ (self.audit / 'independent.txt').write_text('synthetic independent audit fixture')
+ study_inventory = package.tree(self.study)
+ write_json(self.audit / 'audit.json', {
+ 'schema': 'dvidia.retention-independent-audit/v1', 'passed': True,
+ 'source_and_raw_inputs_unchanged': True,
+ 'protocol_sha256': package.identity(self.study / 'protocol.json')['sha256'],
+ 'source_sha256': {}, 'frozen_raw_files': study_inventory, 'rows': rows})
+
+ def test_prepares_repeatable_complete_archive_with_verified_reconstruction(self):
+ outputs = [self.root / 'package-a', self.root / 'package-b']
+ receipts = [package.prepare(self.study, self.audit, output, Path(__file__).resolve().parents[2])
+ for output in outputs]
+ archive_a = outputs[0] / 'huggingface' / package.ARCHIVE_NAME
+ archive_b = outputs[1] / 'huggingface' / package.ARCHIVE_NAME
+ self.assertEqual(archive_a.read_bytes(), archive_b.read_bytes())
+ self.assertEqual(receipts[0]['verified']['verified'], True)
+ self.assertEqual(receipts[0]['attempts'], 36)
+ self.assertEqual({path.name for path in (outputs[0] / 'huggingface').iterdir()}, package.HF_NAMES)
+ with zipfile.ZipFile(archive_a) as archive:
+ self.assertIn('study/report.json', archive.namelist())
+ self.assertIn('audit/audit.json', archive.namelist())
+ self.assertIn('tools/verify_archive.py', archive.namelist())
+ self.assertEqual(json.loads(archive.read('inventory.json'))['schema'],
+ 'dvidia.piper-retention-inventory/v1')
+ verifier = {'__name__': 'retention_archive_verifier'}
+ exec(compile(package.VERIFY_SOURCE, 'verify_archive.py', 'exec'), verifier)
+ destination = self.root / 'reconstructed'
+ proof = verifier['verify'](archive_a, destination)
+ self.assertTrue(proof['verified'])
+ self.assertEqual((destination / 'study/report.json').read_bytes(),
+ (self.study / 'report.json').read_bytes())
+
+ def test_requires_passed_audit_and_same_twelve_scene_roster(self):
+ audit = json.loads((self.audit / 'audit.json').read_bytes())
+ audit['passed'] = False
+ write_json(self.audit / 'audit.json', audit)
+ with self.assertRaisesRegex(ValueError, 'audit'):
+ package.validate_inputs(self.study, self.audit)
+ audit['passed'] = True
+ write_json(self.audit / 'audit.json', audit)
+ report = json.loads((self.study / 'report.json').read_bytes())
+ report['rows'][-1]['scene_id'] = 'different-scene'
+ write_json(self.study / 'report.json', report)
+ with self.assertRaisesRegex(ValueError, 'twelve scenes'):
+ package.validate_inputs(self.study, self.audit)
+
+ def test_refuses_duplicate_nonfinite_and_private_evidence(self):
+ (self.study / 'bad.json').write_bytes(b'{"x":1,"x":2}')
+ with self.assertRaises(ValueError):
+ package.tree(self.study)
+ (self.study / 'bad.json').unlink()
+ (self.study / 'bad.json').write_bytes(b'{"x":NaN}')
+ with self.assertRaises(ValueError):
+ package.tree(self.study)
+ (self.study / 'bad.json').unlink()
+ (self.study / 'attempt-0.bin').write_bytes(b'/'+b'Users/private/person/video.mp4')
+ with self.assertRaisesRegex(ValueError, 'Private path'):
+ package.tree(self.study)
+
+ def test_verifier_rejects_modified_archive_and_refuses_overwrite(self):
+ receipt = package.prepare(self.study, self.audit, self.root / 'package',
+ Path(__file__).resolve().parents[2])
+ archive_path = self.root / 'package/huggingface' / package.ARCHIVE_NAME
+ verifier = {'__name__': 'retention_archive_verifier'}
+ exec(compile(package.VERIFY_SOURCE, 'verify_archive.py', 'exec'), verifier)
+ output = self.root / 'reconstructed'
+ verifier['verify'](archive_path, output)
+ with self.assertRaisesRegex(ValueError, 'new'):
+ verifier['verify'](archive_path, output)
+ altered = self.root / 'altered.zip'
+ with zipfile.ZipFile(archive_path) as source, zipfile.ZipFile(altered, 'w') as dest:
+ for name in source.namelist():
+ raw = source.read(name)
+ if name == 'study/report.json':
+ raw += b' '
+ dest.writestr(name, raw)
+ with self.assertRaisesRegex(ValueError, 'expected bytes|digest'):
+ verifier['verify'](altered)
+ self.assertEqual(receipt['archive']['sha256'], sha256(archive_path.read_bytes()).hexdigest())
+
+ def test_prepare_rejects_transient_same_size_mutation_before_archive_indexing(self):
+ target = self.study/'attempt-0.bin'
+ original = target.read_bytes()
+ real_archive = package.archive
+ def mutate_then_archive(*args, **kwargs):
+ target.write_bytes(b'X'*len(original))
+ try:
+ return real_archive(*args, **kwargs)
+ finally:
+ # Restoration would defeat a final-tree-only check.
+ target.write_bytes(original)
+ with patch.object(package, 'archive', side_effect=mutate_then_archive):
+ with self.assertRaisesRegex(ValueError, 'pinned evidence identity'):
+ package.prepare(self.study,self.audit,self.root/'mutated-index',
+ Path(__file__).resolve().parents[2])
+ self.assertEqual(target.read_bytes(), original)
+
+ def test_archive_stream_copy_rejects_mutation_after_indexing(self):
+ target = self.study/'attempt-0.bin'
+ original, real_zip = target.read_bytes(), zipfile.ZipFile
+ expected = {'study/attempt-0.bin': package.identity(target)}
+ def mutate_before_copy(*args, **kwargs):
+ target.write_bytes(b'Y'*len(original))
+ return real_zip(*args, **kwargs)
+ try:
+ with patch.object(package.zipfile, 'ZipFile', side_effect=mutate_before_copy):
+ with self.assertRaisesRegex(ValueError, 'Copied archive input differs'):
+ package.archive(self.root/'mutated-copy.zip',{'study/attempt-0.bin':target},{},
+ expected_identities=expected)
+ finally:
+ target.write_bytes(original)
+
+ def test_parsed_metadata_is_bound_to_the_tree_snapshot(self):
+ real_load = package.load_json
+ def changed_read(path, *args, **kwargs):
+ value, raw = real_load(path, *args, **kwargs)
+ return (value, raw+b' ') if Path(path).name=='report.json' else (value,raw)
+ with patch.object(package, 'load_json', side_effect=changed_read):
+ with self.assertRaisesRegex(ValueError, 'Parsed metadata differs'):
+ package.validate_inputs(self.study,self.audit)
+
+ def test_archive_identity_has_its_own_larger_byte_bound(self):
+ large = self.root/package.ARCHIVE_NAME
+ with large.open('wb') as stream:
+ stream.truncate(package.MAX_FILE+1)
+ with self.assertRaisesRegex(ValueError, 'oversized'):
+ package.identity(large)
+ self.assertEqual(package.identity(large,package.MAX_ARCHIVE),package.archive_digest(large))
+ real_identity, archive_calls = package.identity, []
+ def checked_identity(path, limit=package.MAX_FILE):
+ if Path(path).name==package.ARCHIVE_NAME:
+ self.assertEqual(limit,package.MAX_ARCHIVE)
+ archive_calls.append(path)
+ return real_identity(path,limit)
+ with patch.object(package, 'identity', side_effect=checked_identity):
+ package.prepare(self.study,self.audit,self.root/'archive-limit',
+ Path(__file__).resolve().parents[2])
+ self.assertEqual(len(archive_calls),1)
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/publishing/huggingface/publish_piper_retention_v02.py b/publishing/huggingface/publish_piper_retention_v02.py
new file mode 100644
index 0000000..79af2ef
--- /dev/null
+++ b/publishing/huggingface/publish_piper_retention_v02.py
@@ -0,0 +1,184 @@
+"""Stage or publish the reviewed PiPER retention v0.2 evidence dataset.
+
+This publisher is intentionally inert until its explicit ``publish`` command.
+It freezes exactly six reviewed files and writes one commit only to an empty,
+public dataset head. It does not modify any Space.
+"""
+from __future__ import annotations
+
+import argparse
+from hashlib import sha256
+import json
+from pathlib import Path
+import re
+import sys
+
+from publish import PublicationError, client
+from publish_task_environments_v02 import digest, encoded, plain, public_client, download, scan
+
+
+REPO = 'Dvidia/piper-retention-v02'
+SCHEMA = 'dvidia.piper-retention-publication/v1'
+ARCHIVE = 'piper-retention-v02.zip'
+NAMES = {'README.md', 'LICENSE', ARCHIVE, 'report.json', 'protocol.json', 'inspection.json'}
+PLAN_LIMIT = 4 * 1024 * 1024
+LIMITS = {name: (192 * 1024 * 1024 if name == ARCHIVE else
+ 16 * 1024 * 1024 if name == 'report.json' else 4 * 1024 * 1024)
+ for name in NAMES}
+OID = re.compile(r'(?:[a-f0-9]{40}|[a-f0-9]{64})\Z')
+
+
+def read(path, limit=None):
+ path = plain(path)
+ limit = LIMITS.get(path.name, PLAN_LIMIT) if limit is None else limit
+ if not path.is_file() or path.stat().st_size > limit:
+ raise PublicationError('Missing or oversized reviewed publication file.')
+ with path.open('rb') as stream:
+ raw = stream.read(limit + 1)
+ if len(raw) > limit:
+ raise PublicationError('Publication file changed beyond its byte limit.')
+ return raw
+
+
+def files(root):
+ root = plain(root)
+ if not root.is_dir():
+ raise PublicationError('Expected a plain publication directory.')
+ paths = sorted(root.iterdir())
+ if {path.name for path in paths} != NAMES:
+ raise PublicationError('Dataset must contain exactly the reviewed six files.')
+ inventory = []
+ for path in paths:
+ raw = read(path)
+ scan(raw)
+ inventory.append({'path': path.name, **digest(raw),
+ 'representation': 'lfs' if path.name == ARCHIVE else 'git'})
+ return inventory
+
+
+def stage(root, plan):
+ root, plan = plain(root), plain(plan)
+ value = {'schema': SCHEMA, 'repo_id': REPO, 'repo_type': 'dataset',
+ 'private': False, 'root': str(root), 'files': files(root)}
+ with plan.open('xb') as stream:
+ stream.write(encoded(value))
+ return {'plan_sha256': sha256(read(plan, PLAN_LIMIT)).hexdigest(),
+ 'files': len(value['files']), 'bytes': sum(row['bytes'] for row in value['files'])}
+
+
+def _pairs(items):
+ result = {}
+ for key, value in items:
+ if key in result:
+ raise PublicationError('Duplicate keys in reviewed publication plan.')
+ result[key] = value
+ return result
+
+
+def _constant(_value):
+ raise PublicationError('Nonfinite values in reviewed publication plan.')
+
+
+def checked(plan, expected_digest):
+ raw = read(plan, PLAN_LIMIT)
+ if sha256(raw).hexdigest() != expected_digest:
+ raise PublicationError('Reviewed plan digest changed.')
+ try:
+ value = json.loads(raw, object_pairs_hook=_pairs, parse_constant=_constant)
+ except (ValueError, TypeError, UnicodeError) as exc:
+ raise PublicationError('Malformed reviewed publication plan.') from exc
+ if (type(value) is not dict
+ or set(value) != {'schema', 'repo_id', 'repo_type', 'private', 'root', 'files'}
+ or value['schema'] != SCHEMA or value['repo_id'] != REPO
+ or value['repo_type'] != 'dataset' or value['private'] is not False
+ or type(value['root']) is not str or type(value['files']) is not list
+ or len(value['files']) != len(NAMES)):
+ raise PublicationError('Reviewed dataset identity changed.')
+ for row in value['files']:
+ if (type(row) is not dict or set(row) != {'path', 'bytes', 'sha256', 'representation'}
+ or type(row['path']) is not str or row['path'] not in NAMES
+ or type(row['bytes']) is not int or row['bytes'] < 0
+ or type(row['sha256']) is not str or not re.fullmatch(r'[a-f0-9]{64}', row['sha256'])
+ or row['representation'] != ('lfs' if row['path'] == ARCHIVE else 'git')):
+ raise PublicationError('Reviewed dataset inventory is malformed.')
+ if ({row['path'] for row in value['files']} != NAMES
+ or files(value['root']) != value['files']):
+ raise PublicationError('Reviewed dataset identity or bytes changed.')
+ return value
+
+
+def _remote(info):
+ if (getattr(info, 'private', None) is not False
+ or not OID.fullmatch(getattr(info, 'sha', '') or '')
+ or not getattr(info, 'siblings', None)):
+ raise PublicationError('Expected an initialized public dataset with an exact commit.')
+
+
+def publish(plan, expected_digest, cache):
+ value = checked(plan, expected_digest)
+ payloads = []
+ for row in value['files']:
+ raw = read(Path(value['root']) / row['path'])
+ if digest(raw) != {key: row[key] for key in ('bytes', 'sha256')}:
+ raise PublicationError('Payload changed while freezing reviewed publication bytes.')
+ payloads.append((row['path'], raw))
+ # No payload file is reopened after credential loading or remote access.
+ api, token = client(cache)
+ from huggingface_hub import CommitOperationAdd
+ from huggingface_hub.errors import RepositoryNotFoundError
+ try:
+ info = api.repo_info(REPO, repo_type='dataset', files_metadata=True)
+ except RepositoryNotFoundError:
+ api.create_repo(REPO, repo_type='dataset', private=False, exist_ok=False)
+ info = api.repo_info(REPO, repo_type='dataset', files_metadata=True)
+ _remote(info)
+ if len(info.siblings) != 1 or {item.rfilename for item in info.siblings} != {'.gitattributes'}:
+ raise PublicationError('Dataset is not empty. Verify the existing publication; do not overwrite it.')
+ operations = [CommitOperationAdd(path_in_repo=name, path_or_fileobj=raw) for name, raw in payloads]
+ commit = api.create_commit(REPO, repo_type='dataset', operations=operations,
+ parent_commit=info.sha,
+ commit_message='Publish PiPER retention v0.2 simulator evidence')
+ del token, api
+ return {'repo_id': REPO, 'commit': commit.oid, 'files_added': len(operations)}
+
+
+def verify(plan, expected_digest, cache):
+ value = checked(plan, expected_digest)
+ api = public_client()
+ info = api.repo_info(REPO, repo_type='dataset', files_metadata=True, token=False)
+ _remote(info)
+ if (len(info.siblings) != len(NAMES) + 1
+ or {item.rfilename for item in info.siblings} != NAMES | {'.gitattributes'}):
+ raise PublicationError('Public dataset roster differs from the reviewed files.')
+ for row in value['files']:
+ raw = download(REPO, row['path'], 'dataset', info.sha, cache)
+ if digest(raw) != {key: row[key] for key in ('bytes', 'sha256')}:
+ raise PublicationError('Anonymous public bytes differ from the reviewed artifacts.')
+ return {'repo_id': REPO, 'commit': info.sha, 'anonymous_files_verified': len(value['files'])}
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__)
+ commands = parser.add_subparsers(dest='command', required=True)
+ command = commands.add_parser('stage')
+ command.add_argument('root'); command.add_argument('--plan', required=True)
+ for name in ('publish', 'verify'):
+ command = commands.add_parser(name)
+ command.add_argument('--plan', required=True)
+ command.add_argument('--sha256', required=True)
+ command.add_argument('--cache', required=True)
+ args = parser.parse_args()
+ try:
+ result = (stage(args.root, args.plan) if args.command == 'stage' else
+ globals()[args.command](args.plan, args.sha256, Path(args.cache)))
+ print(json.dumps(result, indent=2))
+ return 0
+ except PublicationError as exc:
+ print(str(exc), file=sys.stderr)
+ except Exception:
+ print('Publication operation failed; no credential or remote response details displayed.', file=sys.stderr)
+ return 2
+
+
+if __name__ == '__main__':
+ raise SystemExit(main())
diff --git a/publishing/huggingface/test_publish_piper_retention_v02.py b/publishing/huggingface/test_publish_piper_retention_v02.py
new file mode 100644
index 0000000..1c4c20d
--- /dev/null
+++ b/publishing/huggingface/test_publish_piper_retention_v02.py
@@ -0,0 +1,143 @@
+from copy import deepcopy
+from hashlib import sha256
+import json
+from pathlib import Path
+import sys
+from types import SimpleNamespace
+import tempfile
+import unittest
+from unittest.mock import Mock, patch
+
+import publish_piper_retention_v02 as publisher
+
+
+class RetentionPublicationTests(unittest.TestCase):
+ def setUp(self):
+ temporary = tempfile.TemporaryDirectory()
+ self.addCleanup(temporary.cleanup)
+ self.root = Path(temporary.name).resolve()
+ self.dataset = self.root / 'dataset'; self.dataset.mkdir()
+ for name in publisher.NAMES:
+ (self.dataset / name).write_bytes(b'synthetic retention fixture')
+ self.plan = self.root / 'plan.json'
+ self.receipt = publisher.stage(self.dataset, self.plan)
+ class Operation:
+ def __init__(self, path_in_repo, path_or_fileobj):
+ self.path_in_repo, self.path_or_fileobj = path_in_repo, path_or_fileobj
+ class NotFound(Exception):
+ pass
+ sdk = SimpleNamespace(CommitOperationAdd=Operation)
+ errors = SimpleNamespace(RepositoryNotFoundError=NotFound)
+ patcher = patch.dict(sys.modules, {'huggingface_hub': sdk,
+ 'huggingface_hub.errors': errors})
+ patcher.start(); self.addCleanup(patcher.stop)
+
+ def info(self, names=('.gitattributes',), private=False, revision='a' * 40):
+ return SimpleNamespace(private=private, sha=revision,
+ siblings=[SimpleNamespace(rfilename=name) for name in names])
+
+ def test_staging_checking_and_public_readback_do_not_load_credentials(self):
+ api = Mock(); api.repo_info.return_value = self.info(sorted(publisher.NAMES | {'.gitattributes'}))
+ with patch.object(publisher, 'client', side_effect=AssertionError('Credential accessed')), \
+ patch.object(publisher, 'public_client', return_value=api), \
+ patch.object(publisher, 'download', side_effect=lambda repo, name, kind, revision, cache:
+ (self.dataset / name).read_bytes()) as download:
+ value = publisher.checked(self.plan, self.receipt['plan_sha256'])
+ proof = publisher.verify(self.plan, self.receipt['plan_sha256'], self.root / 'cache')
+ self.assertEqual(value['repo_id'], 'Dvidia/piper-retention-v02')
+ self.assertEqual(proof['anonymous_files_verified'], 6)
+ self.assertEqual(download.call_count, 6)
+ api.repo_info.assert_called_once_with(publisher.REPO, repo_type='dataset',
+ files_metadata=True, token=False)
+ self.assertTrue(all(call.args[3] == 'a' * 40 for call in download.call_args_list))
+ with self.assertRaises(FileExistsError):
+ publisher.stage(self.dataset, self.plan)
+
+ def test_mutated_payload_and_unreviewed_roster_fail_before_credentials(self):
+ with patch.object(publisher, 'client') as client:
+ (self.dataset / 'README.md').write_bytes(b'changed')
+ with self.assertRaises(publisher.PublicationError):
+ publisher.publish(self.plan, self.receipt['plan_sha256'], self.root / 'cache')
+ (self.dataset / 'README.md').write_bytes(b'synthetic retention fixture')
+ (self.dataset / 'unreviewed.json').write_text('{}')
+ with self.assertRaises(publisher.PublicationError):
+ publisher.publish(self.plan, self.receipt['plan_sha256'], self.root / 'cache')
+ client.assert_not_called()
+
+ def test_resealed_destination_duplicate_and_nonfinite_plans_are_rejected(self):
+ original = json.loads(self.plan.read_bytes())
+ for field, replacement in [('repo_id', 'Dvidia/robot-simulation-lab'),
+ ('repo_type', 'space'), ('private', True)]:
+ value = deepcopy(original); value[field] = replacement
+ raw = publisher.encoded(value); self.plan.write_bytes(raw)
+ with self.subTest(field=field), self.assertRaises(publisher.PublicationError):
+ publisher.checked(self.plan, sha256(raw).hexdigest())
+ for raw in (b'{"schema":1,"schema":2}', b'{"private":NaN}'):
+ self.plan.write_bytes(raw)
+ with self.assertRaises(publisher.PublicationError):
+ publisher.checked(self.plan, sha256(raw).hexdigest())
+ value = deepcopy(original); value['files'][1] = value['files'][0]
+ raw = publisher.encoded(value); self.plan.write_bytes(raw)
+ with self.assertRaises(publisher.PublicationError):
+ publisher.checked(self.plan, sha256(raw).hexdigest())
+
+ def test_existing_private_or_unpinned_dataset_cannot_be_overwritten(self):
+ for info in (self.info(['README.md']), self.info(private=True), self.info(revision='main')):
+ api = Mock(); api.repo_info.return_value = info
+ with self.subTest(info=info), patch.object(publisher, 'client', return_value=(api, 'fixture')):
+ with self.assertRaises(publisher.PublicationError):
+ publisher.publish(self.plan, self.receipt['plan_sha256'], self.root / 'cache')
+ api.create_commit.assert_not_called()
+ api.create_repo.assert_not_called()
+
+ def test_payload_freeze_and_upload_bytes_are_immutable_across_credential_boundary(self):
+ original = publisher.read
+ reads = 0
+ def changed(path, *args, **kwargs):
+ nonlocal reads
+ if Path(path).name == 'README.md':
+ reads += 1
+ if reads == 2:
+ return b'changed after review'
+ return original(path, *args, **kwargs)
+ with patch.object(publisher, 'read', side_effect=changed), patch.object(publisher, 'client') as client:
+ with self.assertRaisesRegex(publisher.PublicationError, 'freezing'):
+ publisher.publish(self.plan, self.receipt['plan_sha256'], self.root / 'cache')
+ client.assert_not_called()
+ api = Mock(); api.repo_info.return_value = self.info()
+ api.create_commit.return_value = SimpleNamespace(oid='b' * 40)
+ def credential_boundary(_cache):
+ (self.dataset / 'README.md').write_bytes(b'late mutation not uploaded')
+ return api, 'fixture'
+ with patch.object(publisher, 'client', side_effect=credential_boundary):
+ result = publisher.publish(self.plan, self.receipt['plan_sha256'], self.root / 'cache')
+ self.assertEqual(result['files_added'], 6)
+ commit = api.create_commit.call_args
+ self.assertEqual(commit.kwargs['parent_commit'], 'a' * 40)
+ uploaded = {operation.path_in_repo: operation.path_or_fileobj
+ for operation in commit.kwargs['operations']}
+ self.assertEqual(uploaded['README.md'], b'synthetic retention fixture')
+
+ def test_oversized_file_and_changed_anonymous_bytes_are_rejected(self):
+ path = self.dataset / publisher.ARCHIVE
+ with path.open('wb') as stream:
+ stream.truncate(publisher.LIMITS[publisher.ARCHIVE] + 1)
+ with self.assertRaisesRegex(publisher.PublicationError, 'oversized'):
+ publisher.read(path)
+ (self.dataset / publisher.ARCHIVE).write_bytes(b'synthetic retention fixture')
+ self.plan.unlink(); self.receipt = publisher.stage(self.dataset, self.plan)
+ api = Mock(); api.repo_info.return_value = self.info(sorted(publisher.NAMES | {'.gitattributes'}))
+ with patch.object(publisher, 'client', side_effect=AssertionError('Credential accessed')), \
+ patch.object(publisher, 'public_client', return_value=api), \
+ patch.object(publisher, 'download', return_value=b'changed bytes'):
+ with self.assertRaisesRegex(publisher.PublicationError, 'Anonymous public bytes'):
+ publisher.verify(self.plan, self.receipt['plan_sha256'], self.root / 'cache')
+ target = self.root / 'unreviewed'; target.write_text('fixture')
+ (self.dataset / 'README.md').unlink()
+ (self.dataset / 'README.md').symlink_to(target)
+ with self.assertRaises(publisher.PublicationError):
+ publisher.checked(self.plan, self.receipt['plan_sha256'])
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/src/dvidia_training/piper_retention.py b/src/dvidia_training/piper_retention.py
new file mode 100644
index 0000000..9b182ce
--- /dev/null
+++ b/src/dvidia_training/piper_retention.py
@@ -0,0 +1,236 @@
+"""Three bounded, authored native-PiPER holds after the unchanged evidence probe.
+
+All inputs are privileged simulator observations. Position regulation and a
+synthetic force ceiling do not implement physical stopping or force control.
+The fixed parameters below are engineering assumptions, not tuned hardware
+calibration. No full transport, squeeze escalation or requalification occurs.
+"""
+from __future__ import annotations
+
+from copy import deepcopy
+import math
+
+from . import piper_probe as probe
+
+SCHEMA = 'dvidia.piper-retention/v2'
+VARIANTS = ('measured_joint', 'issued_joint', 'cartesian')
+FORCE_CEILING_N = 10.
+POSITION_ERROR_CEILING_M = .004
+ORIENTATION_ERROR_CEILING_RAD = .05
+INITIAL_BIAS_CEILING_RAD = .08
+CORRECTION_CEILING_RAD = .03
+CORRECTION_RATE_RAD_S = .10
+JOINT_MARGIN_RAD = .02
+
+
+def retention_contract():
+ return {
+ 'schema': SCHEMA, 'variants': list(VARIANTS), 'physical_robot_ready': False,
+ 'observation_origin': probe.ORIGIN, 'control_dt_s': probe.DT,
+ 'acquisition': 'Unchanged ProbeStopPolicy evidence acquisition, including its bounded supported retry. All variants share the same prefix through the first admission observation.',
+ 'measured_joint': 'Original hold: latch measured six joint coordinates and the last issued jaw target at admission.',
+ 'issued_joint': 'Latch the preceding issued six joint targets and the same last issued jaw target at admission.',
+ 'cartesian': 'Latch measured TCP position/orientation and preceding issued targets. Fixed joint bias equals issued target minus pose_action at the admission pose. First command repeats the exact issued target; later pose_action updates receive that fixed bias.',
+ 'limits': {'per_finger_force_ceiling_n': FORCE_CEILING_N,
+ 'tcp_position_error_ceiling_m': POSITION_ERROR_CEILING_M,
+ 'tcp_orientation_error_ceiling_rad': ORIENTATION_ERROR_CEILING_RAD,
+ 'initial_joint_bias_ceiling_rad': INITIAL_BIAS_CEILING_RAD,
+ 'joint_correction_ceiling_rad': CORRECTION_CEILING_RAD,
+ 'joint_correction_rate_rad_s': CORRECTION_RATE_RAD_S,
+ 'native_joint_target_margin_rad': JOINT_MARGIN_RAD},
+ 'rate_scope': 'The extra correction slews at at most 0.002 rad per 20 ms per joint. Original native 1 ms target slew, servos, gains and force caps remain unchanged.',
+ 'refusal': 'Original eligibility loss, unexpected contact or observed pad force >10 N terminally requests measured joint hold and the frozen jaw target. Cartesian error, initial-bias, total-correction or native-margin exhaustion also refuses terminally; it never chases an already failed grip.',
+ 'invalid_measurements': 'Malformed, nonfinite or noncausal observations terminally refuse and raise without issuing a command. No measured-position stop is claimed from invalid measurements.',
+ 'final_observation': 'Repeat force, original eligibility and Cartesian pose-error checks. Do not solve or issue a hypothetical next control at the terminal observation.',
+ 'sampling': 'Control checks occur at 20 ms boundaries. Independent saved 1 ms native evidence must judge a continuous 5 s hold; missed between-boundary failures remain findings.',
+ 'qualifications': 'Engineering limits and the 10 N ceiling are authored synthetic assumptions, not damage limits or calibrated physical telemetry. Position targets are not physical stops or force limiters. No load/friction inference, model training or complete placement is implemented.',
+ }
+
+
+class RetentionPolicy(probe.ProbeStopPolicy):
+ """Preserve acquisition while changing only the admitted hold request."""
+
+ def __init__(self, environment, variant='measured_joint'):
+ if type(variant) is not str or variant not in VARIANTS:
+ raise ValueError('Unknown bounded retention variant')
+ super().__init__(environment, variant='evidence')
+ self.retention_variant = variant
+ self.hold_reference, self.hold_records, self.input_refusals = None, [], []
+ self.input_refused = False
+ self.correction = [0.]*6
+ self.calculation = {}
+ self.final_inspection = False
+ ranges = environment.model.actuator_ctrlrange
+ if len(ranges) != 7:
+ raise ValueError('Retention requires the native seven-actuator PiPER model')
+ self.joint_limits = []
+ for row in ranges[:6]:
+ if len(row) != 2:
+ raise ValueError('Malformed native arm control limits')
+ low, high = (probe._number(float(value), 'native joint limit') for value in row)
+ if high-low <= 2*JOINT_MARGIN_RAD:
+ raise ValueError('Native joint interval cannot contain the unchanged margin')
+ self.joint_limits.append([low, high])
+
+ def _clock(self, observation):
+ try:
+ super()._clock(observation)
+ except ValueError:
+ self.input_refused = True
+ self.admitted = self.qualified = False
+ self.aborted, self.state = True, 'abort'
+ self.status, self.abort_reason = 'invalid_measurement_refused', 'invalid_measurement_or_cadence'
+ refusal = {'last_valid_observation_s': self.last_observed_s,
+ 'requested_control_count': len(self.requested_controls),
+ 'reason': self.abort_reason, 'command_issued': False,
+ 'scope': 'No trusted current position exists; no new hold target was issued.'}
+ self.input_refusals.append(refusal)
+ raise
+ if (self.status not in ('invalidated_hold', 'timed_out_hold')
+ and max(observation['pad_normal_forces_n'].values()) > FORCE_CEILING_N):
+ self.diagnostic = self._diagnose(observation)
+ self._terminal_hold(observation, 'declared_force_ceiling_exceeded')
+
+ def _event(self, kind, observation, reason):
+ super()._event(kind, observation, reason)
+ self.events[-1]['retention_variant'] = self.retention_variant
+
+ def _pose_errors(self, observation):
+ if self.hold_reference is None:
+ return None, None
+ reference = self.hold_reference
+ position = math.dist(observation['tcp_position_m'], reference['position_m'])
+ product = abs(sum(a*b for a, b in zip(observation['tcp_quaternion_wxyz'], reference['quaternion_wxyz'])))
+ orientation = 2*math.acos(min(1., product))
+ return position, orientation
+
+ def _range_failure(self, targets, *, margin):
+ return any(not low+margin-1e-12 <= value <= high-margin+1e-12
+ for value, (low, high) in zip(targets, self.joint_limits))
+
+ def _refuse_bound(self, observation, reason):
+ self.calculation['refused_bounds'] = reason
+ return self._terminal_hold(observation, reason)
+
+ def _admit(self, observation, kind):
+ if self.last_requested is None:
+ return self._refuse_bound(observation, 'missing_prior_issued_target')
+ issued = probe._vector(self.last_requested['joint_targets_rad'], 'prior issued targets', 6)
+ jaw = self._jaw_to_freeze(observation)
+ self.hold_reference = {
+ 't_seconds': observation['t_seconds'], 'position_m': list(observation['tcp_position_m']),
+ 'quaternion_wxyz': list(observation['tcp_quaternion_wxyz']),
+ 'issued_joint_targets_rad': issued, 'measured_joint_position_rad': list(observation['joint_position_rad']),
+ 'frozen_jaw_target_m': jaw, 'raw_pose_joint_targets_rad': None, 'fixed_bias_rad': None,
+ }
+ if self.retention_variant == 'cartesian':
+ try:
+ raw = self.env.pose_action(self.hold_reference['position_m'], self.hold_reference['quaternion_wxyz'], jaw)
+ raw = probe._vector(raw['joint_targets_rad'], 'admission pose action', 6)
+ except (ValueError, TypeError, KeyError, ArithmeticError):
+ return self._refuse_bound(observation, 'pose_solver_refused')
+ bias = [target-current for target, current in zip(issued, raw)]
+ self.hold_reference.update(raw_pose_joint_targets_rad=raw, fixed_bias_rad=bias)
+ self.calculation = {'raw_pose_joint_targets_rad': raw, 'desired_correction_rad': [0.]*6,
+ 'correction_rate_limited_axes': [], 'refused_bounds': None}
+ if max(map(abs, bias)) > INITIAL_BIAS_CEILING_RAD+1e-12:
+ return self._refuse_bound(observation, 'initial_bias_exhausted')
+ if self._range_failure(issued, margin=JOINT_MARGIN_RAD):
+ return self._refuse_bound(observation, 'native_joint_margin_exhausted')
+ action = super()._admit(observation, kind)
+ if self.retention_variant != 'measured_joint':
+ # Explicitly reuse the prior vector: raw+bias can differ by rounding.
+ self.stop_targets = list(issued)
+ action = self._hold()
+ self.events[-1]['retention_hold_reference'] = deepcopy(self.hold_reference)
+ self.events[-1]['diagnostic']['decision'] = 'bounded_'+self.retention_variant+'_hold'
+ return action
+
+ def _hold_check(self, observation):
+ action = super()._hold_check(observation)
+ if not self.admitted or self.retention_variant != 'cartesian':
+ return action
+ position_error, orientation_error = self._pose_errors(observation)
+ self.calculation = {'raw_pose_joint_targets_rad': None, 'desired_correction_rad': None,
+ 'correction_rate_limited_axes': [], 'refused_bounds': None}
+ if position_error > POSITION_ERROR_CEILING_M+1e-12:
+ return self._refuse_bound(observation, 'tcp_position_error_exhausted')
+ if orientation_error > ORIENTATION_ERROR_CEILING_RAD+1e-12:
+ return self._refuse_bound(observation, 'tcp_orientation_error_exhausted')
+ if self.final_inspection:
+ return action
+ reference = self.hold_reference
+ try:
+ raw = self.env.pose_action(reference['position_m'], reference['quaternion_wxyz'], self.frozen_jaw_target)
+ raw = probe._vector(raw['joint_targets_rad'], 'retention pose action', 6)
+ except (ValueError, TypeError, KeyError, ArithmeticError):
+ return self._refuse_bound(observation, 'pose_solver_refused')
+ # raw+bias-issued == raw-raw_at_admission; keep the fixed bias exactly.
+ desired = [value-initial for value, initial in zip(raw, reference['raw_pose_joint_targets_rad'])]
+ self.calculation.update(raw_pose_joint_targets_rad=raw, desired_correction_rad=desired)
+ if max(map(abs, desired)) > CORRECTION_CEILING_RAD+1e-12:
+ return self._refuse_bound(observation, 'joint_correction_exhausted')
+ step = CORRECTION_RATE_RAD_S*probe.DT
+ limited = [index for index, (wanted, prior) in enumerate(zip(desired, self.correction))
+ if abs(wanted-prior) > step+1e-12]
+ correction = [prior+max(-step, min(step, wanted-prior)) for wanted, prior in zip(desired, self.correction)]
+ targets = [baseline+value for baseline, value in zip(reference['issued_joint_targets_rad'], correction)]
+ self.calculation['correction_rate_limited_axes'] = limited
+ if self._range_failure(targets, margin=JOINT_MARGIN_RAD):
+ return self._refuse_bound(observation, 'native_joint_margin_exhausted')
+ self.correction, self.stop_targets = correction, targets
+ self.diagnostic['decision'] = 'bounded_cartesian_pose_regulation'
+ return self._hold()
+
+ def _record_hold(self, observation, action, *, final=False):
+ if self.hold_reference is None and self.status not in ('invalidated_hold', 'timed_out_hold'):
+ return
+ reference = self.hold_reference or {}
+ position_error, orientation_error = self._pose_errors(observation)
+ self.hold_records.append({
+ 't_seconds': observation['t_seconds'], 'final_observation': final, 'command_issued': not final,
+ 'status': self.status, 'admitted': self.admitted, 'variant': self.retention_variant,
+ 'reference_position_m': reference.get('position_m'), 'reference_quaternion_wxyz': reference.get('quaternion_wxyz'),
+ 'baseline_issued_joint_targets_rad': reference.get('issued_joint_targets_rad'),
+ 'joint_targets_rad': list(action['joint_targets_rad']) if action is not None else None,
+ 'frozen_jaw_target_m': self.frozen_jaw_target,
+ 'raw_pose_joint_targets_rad': deepcopy(self.calculation.get('raw_pose_joint_targets_rad')),
+ 'fixed_bias_rad': reference.get('fixed_bias_rad'),
+ 'desired_correction_rad': deepcopy(self.calculation.get('desired_correction_rad')),
+ 'correction_rad': list(self.correction) if self.admitted and self.retention_variant == 'cartesian' else None,
+ 'correction_rate_limited': bool(self.calculation.get('correction_rate_limited_axes')),
+ 'correction_rate_limited_axes': list(self.calculation.get('correction_rate_limited_axes', [])),
+ 'tcp_position_error_m': position_error, 'tcp_orientation_error_rad': orientation_error,
+ 'refused_bounds': self.calculation.get('refused_bounds'), 'terminal_reason': self.abort_reason,
+ })
+
+ def __call__(self, observation):
+ if self.input_refused:
+ raise ValueError('Retention input refusal is terminal; no further command may be issued')
+ # A stopped boundary must not inherit a preceding boundary's IK solve
+ # or rate-limiting flags and present them as current computations.
+ self.calculation = {}
+ action = super().__call__(observation)
+ self._record_hold(observation, action)
+ return action
+
+ def observe_final(self, observation):
+ if self.input_refused:
+ raise ValueError('Retention input refusal is terminal; final state is unknown')
+ self.calculation = {}
+ self.final_inspection = True
+ try:
+ super().observe_final(observation)
+ finally:
+ self.final_inspection = False
+ self._record_hold(observation, self.last_requested, final=True)
+ return self.summary()
+
+ def summary(self):
+ result = super().summary()
+ result.update(schema=SCHEMA, variant=self.retention_variant, acquisition_variant='evidence',
+ acquisition_contract=result['contract'], contract=retention_contract())
+ result['retention'] = deepcopy({'variant': self.retention_variant, 'hold_reference': self.hold_reference,
+ 'hold_records': self.hold_records, 'input_refusals': self.input_refusals,
+ 'native_joint_control_limits_rad': self.joint_limits})
+ return result
diff --git a/src/dvidia_training/piper_retention_benchmark.py b/src/dvidia_training/piper_retention_benchmark.py
new file mode 100644
index 0000000..0fa8209
--- /dev/null
+++ b/src/dvidia_training/piper_retention_benchmark.py
@@ -0,0 +1,602 @@
+"""Fixed fresh-scene comparison of authored PiPER short-pause hold controls.
+
+This compares three closed-loop controllers, not learned policies or hardware.
+Every attempt, refusal and native row is retained. The primary endpoint is a
+continuous five-second native predicate after the first valid admission;
+controller status at that endpoint and the final hold are separate outcomes.
+"""
+from __future__ import annotations
+
+import argparse
+from copy import deepcopy
+from dataclasses import asdict, replace
+from datetime import datetime, timezone
+import gzip
+from hashlib import sha256
+import io
+import json
+from pathlib import Path
+import sys
+import time
+import zipfile
+
+import numpy as np
+
+from . import piper_probe_benchmark as prior
+
+SCHEMA = 'dvidia.piper-retention-study/v1'
+VARIANTS = ('measured_joint', 'issued_joint', 'cartesian')
+ORDERS = ((0, 1, 2), (1, 2, 0), (2, 0, 1), (2, 1, 0), (1, 0, 2), (0, 2, 1))
+HOLD_S = 5.
+FORCE_CEILING_N = 10.
+TCP_POSITION_CEILING_M = .004
+TCP_ORIENTATION_CEILING_RAD = .05
+BOUND_REFUSALS = frozenset(('initial_bias_exhausted', 'native_joint_margin_exhausted',
+ 'tcp_position_error_exhausted', 'tcp_orientation_error_exhausted', 'joint_correction_exhausted',
+ 'pose_solver_refused', 'missing_prior_issued_target'))
+MAX_FILE = 32*1024*1024
+SCOPE = ('Authored synthetic PiPER probe and short-pause controls with privileged simulator state. '
+ 'Fresh finite development cells; no learning, full placement, physical sensing or hardware qualification.')
+encoded, digest, write, read = prior.encoded, prior.digest, prior.write, prior.read
+
+
+def case_definitions():
+ """Pure fixed cells/order; no native imports, asset fetch or parameter search."""
+ from .piper_task import PiperTaskConfig
+ base, cases = PiperTaskConfig(), []
+ for mass in (.045, .055):
+ for friction in (.75, .50, .20):
+ for layout, start in enumerate(((.303, -.026), (.307, -.037))):
+ index = len(cases)
+ config = replace(base, mass_kg=mass, friction=friction, object_position_m=(*start, .082))
+ cases.append({'id': f'm{round(mass*1000):02}-f{round(friction*100):02}-p{layout}',
+ 'seed': 311100+index, 'config': asdict(config),
+ 'variant_order': [VARIANTS[i] for i in ORDERS[index % len(ORDERS)]]})
+ return cases
+
+
+def roster(protocol):
+ return [(case, variant) for case in protocol['cases'] for variant in case['variant_order']]
+
+
+def make_protocol(assets):
+ from .robot_profiles import verify_assets, LOCK_PATH
+ from .piper_retention import retention_contract
+ return {'schema': SCHEMA, 'scope': SCOPE, 'physical_robot_ready': False, 'learned_policy': False,
+ 'cohort': 'fresh_development_method_check', 'cases': case_definitions(), 'variants': list(VARIANTS),
+ 'planned_attempts': 36, 'source_sha256': prior.sources(), 'runtime': prior.runtime(),
+ 'asset_provenance': verify_assets('piper', assets),
+ 'asset_lock_sha256': sha256(LOCK_PATH.read_bytes()).hexdigest(),
+ 'controller_contract': retention_contract(), 'force_ceiling_n': FORCE_CEILING_N,
+ 'force_ceiling_scope': 'Unchanged arbitrary synthetic admission ceiling, not measured damage calibration or a force command.',
+ 'primary_endpoint': {'duration_s': HOLD_S, 'native_dt_s': .001,
+ 'definition': 'Correct 100 ms native admission prefix, then every native predicate sample through the next 5 s; complete valid warning-free run, no unexpected contact anywhere in the run and no force-ceiling violation on any subsequent hold row.',
+ 'predicate': 'Both contact counts positive, both pad forces strictly >0.01 N and <=10 N including the admission-prefix window, no table support or unexpected contact, rise >=0.006 m from this probe origin and TCP-local drift <=0.003 m; distance epsilon1e-12.',
+ 'missing_followup': 'No admission or less than five seconds of recorded followup is retained as failure, never padded.',
+ 'controller_status': 'Controller retained-at-5s and final hold are reported separately; native success never grants actuator permission.'},
+ 'secondary_endpoint': {'name': 'grip_retention_and_tcp_stationarity',
+ 'duration_s': HOLD_S, 'position_ceiling_m': TCP_POSITION_CEILING_M,
+ 'orientation_ceiling_rad': TCP_ORIENTATION_CEILING_RAD,
+ 'definition': 'Primary native grip retention succeeds, and every native TCP pose from admission through five seconds stays within 4 mm and 0.05 rad of the admission pose. Identical bounds for all three controllers, measured from recorded native poses.'},
+ 'order': 'Six balanced three-variant permutations repeated twice; each variant occupies each position four times. One attempt per cell/variant; no automatic resume, retry of a trial, selection or tuning.',
+ 'comparison': 'Shared acquisition native prefixes must be exactly equal through the earliest common treatment boundary: an admission or an explicit pre-admission hold-bound refusal in any variant. Commands strictly before that boundary must match. If no treatment begins, compare the full traces. Closed-loop hold controls may then change trajectories. These finite cells do not estimate population accuracy or isolate arbitrary physical mechanisms.',
+ 'timing': 'Unpaced shared-host policy-call and full control-step wall/CPU compute; step includes dense telemetry. Reset/compile and evidence serialization are separate from step samples. No sensor/hardware latency or real-time guarantee.'}
+
+
+def prepare(output, assets):
+ output = Path(output)
+ if output.exists() or output.is_symlink():
+ raise ValueError('Choose a fresh output directory; no evidence is overwritten.')
+ protocol = make_protocol(assets)
+ output.mkdir(parents=True)
+ write(output/'protocol.json', protocol)
+ write(output/'frozen.json', {'created_at_utc': datetime.now(timezone.utc).isoformat(),
+ 'protocol_sha256': sha256((output/'protocol.json').read_bytes()).hexdigest()})
+ sources = output/'sources'; sources.mkdir()
+ for name, expected in protocol['source_sha256'].items():
+ raw = Path(__file__).with_name(name).read_bytes()
+ if sha256(raw).hexdigest() != expected:
+ raise ValueError('Source changed during preparation.')
+ (sources/name).write_bytes(raw)
+ return protocol
+
+
+def _timing(values):
+ result = prior._timing(values)
+ if result is not None:
+ result['p99_s'] = float(np.percentile(values, 99))
+ return result
+
+
+def _attempt(assets, case, variant):
+ from .piper_task import PiperTaskConfig
+ from .piper_feedback_env import FeedbackPiperTaskEnv
+ from .piper_retention import RetentionPolicy
+ from .reference_sim import model_digest
+ env = FeedbackPiperTaskEnv(assets, PiperTaskConfig(**case['config']))
+ result = {'schema': SCHEMA, 'id': case['id']+'-'+variant, 'scene_id': case['id'], 'variant': variant,
+ 'seed': case['seed'], 'config': case['config'], 'valid': False, 'physical_robot_ready': False,
+ 'trajectory': [], 'actions': [], 'error': None, 'metrics': {}, 'policy': None}
+ timers = {name: [] for name in ('policy_wall', 'policy_cpu', 'step_wall', 'step_cpu')}
+ policy = None
+ began = time.perf_counter()
+ try:
+ start = time.perf_counter()
+ obs, info = env.reset(case['seed'])
+ result['setup_reset_wall_s'] = time.perf_counter()-start
+ result.update(initial_object_position_m=env.initial_object_position.tolist(),
+ compiled_model_sha256=model_digest(env.model),
+ native_force_limits=env.model.actuator_forcerange.tolist())
+ policy = RetentionPolicy(env, variant=variant)
+ result['trajectory'].append(obs)
+ for _ in range(round(env.config.timeout_s/env.config.control_dt)):
+ wall, cpu = time.perf_counter(), time.process_time()
+ action = policy(obs)
+ timers['policy_wall'].append(time.perf_counter()-wall)
+ timers['policy_cpu'].append(time.process_time()-cpu)
+ record = {'t_seconds': obs['t_seconds'], **action}
+ result['actions'].append(record)
+ obs.update(command=record, issued_joint_targets_rad=list(action['joint_targets_rad']),
+ issued_gripper_position_m=action['gripper_position_m'])
+ wall, cpu = time.perf_counter(), time.process_time()
+ obs, info = env.step(action)
+ timers['step_wall'].append(time.perf_counter()-wall)
+ timers['step_cpu'].append(time.process_time()-cpu)
+ if obs is not None:
+ result['trajectory'].append(obs)
+ if env.done:
+ break
+ if obs is not None:
+ policy.observe_final(obs)
+ result.update(valid=info['valid'], metrics=info, policy=policy.summary())
+ except Exception as exc:
+ result['error'] = {'type': type(exc).__name__, 'message': str(exc)[:300]}
+ if env.ready:
+ result['metrics'] = env.info()
+ if policy is not None:
+ try:
+ result['policy'] = policy.summary()
+ except Exception as summary_exc:
+ result['policy_summary_error'] = {'type': type(summary_exc).__name__,
+ 'message': str(summary_exc)[:300]}
+ result['timing'] = {'attempt_wall_s': time.perf_counter()-began,
+ 'summaries': {name: _timing(values) for name, values in timers.items()}, 'samples': timers}
+ return result, env.native_arrays(), env.native_metadata()
+
+
+def _index(timestamp, times):
+ index = round(timestamp/.001)-1
+ if not 0 <= index < len(times) or abs(times[index]-timestamp) > 1e-8:
+ raise ValueError('Expected an observed native timestamp.')
+ return index
+
+
+def probe_series(arrays, probe, force_ceiling_n=FORCE_CEILING_N):
+ """Pure causal native predicate; independent of controller status labels."""
+ times = arrays['t_seconds']
+ origin = _index(probe['start_time_s'], times)
+ if (np.linalg.norm(arrays['object_position_m'][origin]-probe['object_origin_m']) > 1e-9
+ or np.linalg.norm(arrays['object_minus_tcp_local_m'][origin]-probe['local_offset_origin_m']) > 1e-9):
+ raise ValueError('Probe origin differs from native evidence.')
+ rise = arrays['object_position_m'][:, 2]-probe['object_origin_m'][2]
+ drift = np.linalg.norm(arrays['object_minus_tcp_local_m']-probe['local_offset_origin_m'], axis=1)
+ condition = ((np.arange(len(times)) > origin)
+ & np.all(arrays['grasp_contact_counts'] > 0, axis=1)
+ & np.all(arrays['pad_normal_forces_n'] > .01, axis=1)
+ & np.all(arrays['pad_normal_forces_n'] <= force_ceiling_n, axis=1)
+ & (arrays['table_object_contact_count'] == 0)
+ & (arrays['unexpected_contact_count'] == 0)
+ & (rise+1e-12 >= .006) & (drift <= .003+1e-12))
+ streak = np.empty(len(times), dtype=np.int64)
+ count = 0
+ for index, good in enumerate(condition):
+ count = count+1 if good else 0
+ streak[index] = count
+ return condition, streak, streak >= 100, rise, drift
+
+
+def _decomposition(arrays, start, end):
+ obj = arrays['object_position_m'][end]-arrays['object_position_m'][start]
+ tcp = arrays['tcp_position_m'][end]-arrays['tcp_position_m'][start]
+ return {'object_world_z_delta_m': float(obj[2]), 'tcp_world_z_delta_m': float(tcp[2]),
+ 'relative_world_z_delta_m': float(obj[2]-tcp[2]),
+ 'local_offset_change_m': float(np.linalg.norm(arrays['object_minus_tcp_local_m'][end]-arrays['object_minus_tcp_local_m'][start])),
+ 'elapsed_s': (end-start)*.001,
+ 'scope': 'Kinematic components, not causal attribution or calibrated physical slip.'}
+
+
+def _stationarity(arrays, start, end, enough, grip_success):
+ positions = arrays['tcp_position_m'][start:end+1]
+ quaternions = arrays['tcp_quaternion_wxyz'][start:end+1]
+ displacement = np.linalg.norm(positions-positions[0], axis=1)
+ # q and -q are the same orientation; shortest quaternion angular distance.
+ angles = 2*np.arccos(np.clip(np.abs(quaternions @ quaternions[0]), 0., 1.))
+ max_position, max_angle = float(np.max(displacement)), float(np.max(angles))
+ within = max_position <= TCP_POSITION_CEILING_M+1e-12 and max_angle <= TCP_ORIENTATION_CEILING_RAD+1e-12
+ return {'required_s': HOLD_S, 'complete_5s': bool(enough),
+ 'position_ceiling_m': TCP_POSITION_CEILING_M,
+ 'orientation_ceiling_rad': TCP_ORIENTATION_CEILING_RAD,
+ 'maximum_tcp_position_excursion_m': max_position,
+ 'maximum_tcp_orientation_excursion_rad': max_angle,
+ 'terminal_tcp_position_delta_m': (positions[-1]-positions[0]).tolist(),
+ 'terminal_tcp_orientation_delta_rad': float(angles[-1]),
+ 'tcp_pose_within_bounds': bool(enough and within),
+ 'grip_retention_and_stationarity': bool(grip_success and enough and within),
+ 'scope': 'Common recorded native TCP pose endpoint, not calibrated hardware stationarity.'}
+
+
+def _corrections(policy):
+ retention = (policy or {}).get('retention') or {}
+ records = retention.get('hold_records', [])
+ flags, refusals = {}, {}
+ active, desired, axis_count = [], [], 0
+ for record in records:
+ for key, value in record.items():
+ if type(value) is bool and ('saturat' in key or 'limit' in key or 'refus' in key):
+ flags[key] = flags.get(key, 0)+int(value)
+ if record.get('refused_bounds') is not None:
+ reason = record['refused_bounds']
+ refusals[reason] = refusals.get(reason, 0)+1
+ axis_count += len(record.get('correction_rate_limited_axes') or [])
+ for key, destination in (('correction_rad', active), ('desired_correction_rad', desired)):
+ if record.get(key) is not None:
+ vector = np.asarray(record[key], dtype=float)
+ if vector.shape != (6,) or not np.isfinite(vector).all():
+ raise ValueError('Malformed correction diagnostic.')
+ destination.append(vector)
+ return {'hold_record_count': len(records), 'flag_counts': flags,
+ 'bound_refusal_counts': refusals, 'rate_limited_axis_occurrences': axis_count,
+ 'maximum_abs_active_correction_rad': float(np.max(np.abs(active))) if active else None,
+ 'maximum_abs_desired_correction_rad': float(np.max(np.abs(desired))) if desired else None,
+ 'input_refusals': deepcopy(retention.get('input_refusals', [])),
+ 'scope': 'Includes recorded final diagnostic checks; only command_issued rows represent issued controls.'}
+
+
+def evaluate(result, arrays, metadata, protocol):
+ """Retain invalid/refused attempts, with their primary endpoint false."""
+ from .probe_evidence import _arrays
+ base = {'status': 'refused', 'primary': {'required_s': HOLD_S, 'complete_5s': False,
+ 'native_pause_success': False, 'success': False, 'controller_retained_at_5s': False,
+ 'combined_qualified_pause': False, 'reason': 'invalid_or_incomplete_attempt'},
+ 'first_admission': None, 'probes': [], 'final': {'controller_qualified': False,
+ 'native_qualified': False, 'both_qualified': False, 'evidence_admissible': False},
+ 'stationarity': {'complete_5s': False, 'tcp_pose_within_bounds': False,
+ 'grip_retention_and_stationarity': False},
+ 'decomposition': None, 'corrections': _corrections(result.get('policy')),
+ 'physical_robot_ready': False}
+ try:
+ data, array_digest = _arrays(arrays, metadata)
+ except (ValueError, TypeError, KeyError) as exc:
+ base['reason'] = 'native_sidecar_refused: '+str(exc)
+ return base
+ metrics, policy = result['metrics'], result.get('policy')
+ complete = (len(result['actions']) == 1000 and len(result['trajectory']) == 1001
+ and len(data['t_seconds']) == 20000)
+ critical = int(np.count_nonzero(data['unexpected_contact_count']))
+ max_force = float(np.max(data['pad_normal_forces_n']))
+ native_valid = (result['valid'] and result['error'] is None and metrics.get('valid') is True
+ and not metrics.get('warnings') and complete
+ and critical == 0 and critical == metrics.get('unexpected_contact_native_ticks'))
+ base['native'] = {'arrays_sha256': array_digest, 'rows': len(data['t_seconds']), 'complete': complete,
+ 'valid': bool(native_valid), 'warnings': metrics.get('warnings'),
+ 'unexpected_contact_native_ticks': critical, 'maximum_per_finger_force_n': max_force}
+ if not native_valid or policy is None:
+ return base
+ times, by_probe = data['t_seconds'], {}
+ for probe in policy['probes']:
+ condition, streak, qualified, rise, drift = probe_series(data, probe, protocol['force_ceiling_n'])
+ by_probe[probe['probe_index']] = (probe, condition, streak, qualified, rise, drift)
+ events = [event for event in policy['events'] if event['type'] == 'evidence_qualified'
+ and event['probe_index'] == probe['probe_index']]
+ claims = [{'t_seconds': event['t_seconds'], 'native_prefix_qualified': bool(qualified[_index(event['t_seconds'], times)])}
+ for event in events]
+ base['probes'].append({'probe_index': probe['probe_index'], 'start_time_s': probe['start_time_s'],
+ 'disposition': probe['disposition'], 'claims': claims,
+ 'final_native_qualified_from_origin': bool(qualified[-1])})
+ final_qualified = bool(by_probe[policy['probes'][-1]['probe_index']][3][-1]) if policy['probes'] else False
+ base['final'].update(controller_qualified=bool(policy['admitted']), native_qualified=final_qualified,
+ both_qualified=bool(policy['admitted'] and final_qualified))
+ try:
+ base['receipt'] = prior._evaluate(result, data, metadata, protocol)
+ base['final']['evidence_admissible'] = bool((base['receipt'].get('verification') or {}).get('admissible'))
+ except (ValueError, TypeError, KeyError) as exc:
+ base['receipt'] = {'status': 'audit_refused', 'reason': str(exc)[:300]}
+ admissions = [event for event in policy['events'] if event['type'] == 'evidence_qualified']
+ base['status'] = 'inspected'
+ if not admissions:
+ base['primary']['reason'] = 'no_admission'
+ return base
+ event = admissions[0]
+ probe, condition, streak, qualified, rise, drift = by_probe[event['probe_index']]
+ start = _index(event['t_seconds'], times)
+ target = event['t_seconds']+HOLD_S
+ enough = target <= times[-1]+1e-8
+ end = _index(target, times) if enough else len(times)-1
+ correct = bool(qualified[start])
+ continuous = bool(np.all(condition[start:end+1]))
+ # Every actual hold row must respect the unchanged synthetic ceiling,
+ # including rows after the primary short-pause endpoint.
+ ceiling_ok = bool(np.all(data['pad_normal_forces_n'][start:] <= protocol['force_ceiling_n']))
+ checks = [check for check in policy['qualification_checks'] if abs(check['t_seconds']-target) <= 1e-8
+ and check['probe_index'] == event['probe_index']]
+ retained = enough and len(checks) == 1 and checks[0]['qualified'] is True
+ success = correct and enough and continuous and ceiling_ok
+ reason = ('false_initial_admission' if not correct else 'insufficient_followup' if not enough else
+ 'native_predicate_lost' if not continuous else 'force_ceiling_exceeded' if not ceiling_ok else 'success')
+ base['first_admission'] = {'time_s': event['t_seconds'], 'probe_index': event['probe_index'],
+ 'native_prefix_qualified': correct, 'native_qualifying_window_s': int(streak[start])*.001}
+ failures = np.flatnonzero(~condition[start:end+1])
+ base['primary'] = {'required_s': HOLD_S, 'target_time_s': target,
+ 'followup_available_s': float(times[-1]-event['t_seconds']), 'complete_5s': bool(enough),
+ 'native_pause_success': bool(success), 'success': bool(success),
+ 'controller_retained_at_5s': bool(retained),
+ 'combined_qualified_pause': bool(success and retained),
+ 'all_native_hold_forces_within_ceiling': ceiling_ok,
+ 'first_native_predicate_failure_s': float(times[start+int(failures[0])]) if failures.size else None,
+ 'reason': reason}
+ base['decomposition'] = _decomposition(data, start, end)
+ base['stationarity'] = _stationarity(data, start, end, enough, success)
+ return base
+
+
+def _compact_policy(policy):
+ if policy is None:
+ return None
+ result = {key: deepcopy(value) for key, value in policy.items()
+ if key not in ('requested_controls', 'qualification_checks')}
+ if 'retention' in result:
+ result['retention'].pop('hold_records', None)
+ return result
+
+
+def aggregates(rows):
+ result = {}
+ for variant in VARIANTS:
+ selected = [row for row in rows if row['variant'] == variant]
+ admissions = [row['evaluation']['first_admission'] for row in selected if row['evaluation']['first_admission'] is not None]
+ result[variant] = {'attempts': len(selected), 'valid': sum(row['valid'] for row in selected),
+ 'admissions': len(admissions),
+ 'false_initial_admissions': sum(not row['native_prefix_qualified'] for row in admissions),
+ 'native_5s_pause_success': sum(row['evaluation']['primary']['success'] for row in selected),
+ 'controller_retained_at_5s': sum(row['evaluation']['primary']['controller_retained_at_5s'] for row in selected),
+ 'combined_qualified_5s_pause': sum(row['evaluation']['primary']['combined_qualified_pause'] for row in selected),
+ 'grip_retention_and_stationarity': sum(row['evaluation']['stationarity']['grip_retention_and_stationarity'] for row in selected),
+ 'final_both_qualified': sum(row['evaluation']['final']['both_qualified'] for row in selected)}
+ return result
+
+
+def run(output, assets):
+ output = Path(output)
+ protocol = read(output/'protocol.json')
+ if encoded(protocol) != encoded(make_protocol(assets)):
+ raise ValueError('Frozen protocol/source/runtime/assets changed; create a new study.')
+ write(output/'started.json', {'protocol_sha256': sha256((output/'protocol.json').read_bytes()).hexdigest(),
+ 'started_at_utc': datetime.now(timezone.utc).isoformat()})
+ rows = []
+ for case, variant in roster(protocol):
+ if prior.sources() != protocol['source_sha256']:
+ raise ValueError('Source changed during the fixed study.')
+ folder = output/(case['id']+'-'+variant); folder.mkdir()
+ write(folder/'started.json', {'scene_id': case['id'], 'variant': variant,
+ 'protocol_sha256': sha256((output/'protocol.json').read_bytes()).hexdigest()})
+ result, arrays, metadata = _attempt(assets, case, variant)
+ (folder/'episode.json.gz').write_bytes(gzip.compress(encoded(result), mtime=0))
+ with (folder/'native.npz').open('xb') as stream:
+ np.savez_compressed(stream, **arrays)
+ write(folder/'native.json', metadata)
+ try:
+ evaluation = evaluate(result, arrays, metadata, protocol)
+ except Exception as exc:
+ evaluation = {'status': 'audit_refused', 'reason': type(exc).__name__+': '+str(exc)[:300],
+ 'primary': {'success': False, 'controller_retained_at_5s': False,
+ 'combined_qualified_pause': False},
+ 'first_admission': None, 'final': {'both_qualified': False}, 'physical_robot_ready': False}
+ evaluation['stationarity'] = {'grip_retention_and_stationarity': False}
+ write(folder/'evaluation.json', evaluation)
+ files = {path.name: {'bytes': path.stat().st_size, 'sha256': sha256(path.read_bytes()).hexdigest()}
+ for path in sorted(folder.iterdir())}
+ row = {'id': result['id'], 'scene_id': case['id'], 'variant': variant,
+ 'mass_kg': case['config']['mass_kg'], 'friction': case['config']['friction'],
+ 'valid': result['valid'], 'error': result['error'], 'metrics': result['metrics'],
+ 'policy': _compact_policy(result.get('policy')), 'timing': result['timing']['summaries'],
+ 'evaluation': evaluation, 'files': files}
+ rows.append(row)
+ write(folder/'summary.json', row)
+ print(json.dumps({'attempt': result['id'], 'valid': result['valid'], 'primary': evaluation['primary']['success']}), flush=True)
+ if prior.sources() != protocol['source_sha256']:
+ raise ValueError('Source changed during an attempt; all evidence remains retained.')
+ if prior.runtime() != protocol['runtime']:
+ raise ValueError('Runtime changed during the study.')
+ report = {'schema': SCHEMA, 'scope': SCOPE, 'planned_attempts': 36, 'attempted': len(rows),
+ 'valid_attempts': sum(row['valid'] for row in rows), 'rows': rows, 'aggregates': aggregates(rows),
+ 'physical_robot_ready': False, 'source_sha256': protocol['source_sha256'],
+ 'previous_learned_heldout_result': '5/12 unchanged. No learning or full-placement evaluation in this study.'}
+ write(output/'report.json', report)
+ return report
+
+
+def _read_native(path):
+ with zipfile.ZipFile(path) as archive:
+ entries = archive.infolist()
+ if (len(entries) != 22 or len({entry.filename for entry in entries}) != 22
+ or sum(entry.file_size for entry in entries) > MAX_FILE
+ or any(entry.file_size > 2*1024*1024 or '/' in entry.filename or not entry.filename.endswith('.npy') for entry in entries)):
+ raise ValueError('Native NPZ exceeds the fixed numeric contract.')
+ with np.load(path, allow_pickle=False) as native:
+ return {name: native[name] for name in native.files}
+
+
+def _controls(episode, arrays):
+ actions, trajectory = episode['actions'], episode['trajectory']
+ if len(actions) != 1000 or len(trajectory) != 1001 or len(arrays['t_seconds']) != 20000:
+ raise ValueError('Complete control reconstruction requires 1000 actions/20000 native rows.')
+ requested = episode['policy']['requested_controls']
+ if len(requested) != 1000:
+ raise ValueError('Missing policy control records.')
+ control = np.array(trajectory[0]['native_actuator_controls'], dtype=float)
+ maximum = np.array([episode['config']['joint_target_rate_rad_s']*.001]*6+[episode['config']['jaw_target_rate_m_s']*.001])
+ for i, (action, request) in enumerate(zip(actions, requested)):
+ if (abs(action['t_seconds']-i*.02) > 1e-8 or request['index'] != i
+ or any(action[key] != request[key] for key in ('t_seconds', 'joint_targets_rad', 'gripper_position_m'))
+ or trajectory[i]['command'] != action
+ or trajectory[i]['issued_joint_targets_rad'] != action['joint_targets_rad']
+ or trajectory[i]['issued_gripper_position_m'] != action['gripper_position_m']):
+ raise ValueError('Issued action/boundary/policy records disagree.')
+ target = np.array(action['joint_targets_rad']+[action['gripper_position_m']])
+ for tick in range(i*20, (i+1)*20):
+ control += np.clip(target-control, -maximum, maximum)
+ actual = np.r_[arrays['arm_actuator_controls_rad'][tick], arrays['applied_jaw_target_m'][tick]]
+ if (not np.allclose(control, actual, rtol=0, atol=1e-12)
+ or arrays['requested_jaw_target_m'][tick] != action['gripper_position_m']):
+ raise ValueError('Native applied targets differ from unchanged slew law.')
+ for field in ('object_position_m', 'joint_position_rad', 'tcp_position_m'):
+ if not np.allclose(trajectory[i+1][field], arrays[field][(i+1)*20-1], rtol=0, atol=1e-9):
+ raise ValueError('Boundary state differs from native record.')
+ if any(trajectory[-1][key] is not None for key in ('command', 'issued_joint_targets_rad', 'issued_gripper_position_m')):
+ raise ValueError('Terminal observation contains an unexecuted command.')
+
+
+def _treatment_boundary(policy):
+ events = (policy or {}).get('events', [])
+ admissions = [event for event in events if event['type'] == 'evidence_qualified']
+ admission = min((event['t_seconds'] for event in admissions), default=None)
+ candidates = []
+ for event in events:
+ if event['type'] == 'evidence_qualified':
+ reason = 'admission'
+ elif (event['type'] == 'hold_invalidated' and event.get('reason') in BOUND_REFUSALS
+ and (admission is None or event['t_seconds'] <= admission+1e-8)):
+ reason = 'pre_admission_bound_refusal'
+ else:
+ continue
+ candidates.append({'time_s': event['t_seconds'], 'reason': reason, 'detail': event.get('reason')})
+ return min(candidates, key=lambda row: row['time_s']) if candidates else None
+
+
+def _common_cutoff(rows):
+ boundaries = []
+ for row in rows:
+ boundary = _treatment_boundary(row.get('policy'))
+ if boundary is not None:
+ boundaries.append({'variant': row['variant'], **boundary})
+ boundaries.sort(key=lambda row: row['variant'])
+ timestamp = min((row['time_s'] for row in boundaries), default=None)
+ reasons = sorted({row['reason'] for row in boundaries if abs(row['time_s']-timestamp) <= 1e-8}) if timestamp is not None else []
+ return {'time_s': timestamp, 'reason': '+'.join(reasons) if reasons else 'full_run_no_treatment',
+ 'treatment_boundaries': boundaries}
+
+
+def _prefix(episode, arrays, *, cutoff=None):
+ if cutoff is None:
+ cutoff = _common_cutoff([episode])
+ timestamp = cutoff['time_s']
+ count = _index(timestamp, arrays['t_seconds'])+1 if timestamp is not None else len(arrays['t_seconds'])
+ digest_native = sha256()
+ for name in sorted(arrays):
+ value = np.ascontiguousarray(arrays[name][:count])
+ digest_native.update(encoded({'name': name, 'shape': list(value.shape), 'dtype': str(value.dtype)}))
+ digest_native.update(value.tobytes())
+ actions = [row for row in episode['actions'] if timestamp is None or row['t_seconds'] < timestamp-1e-8]
+ return {'cutoff_time_s': timestamp, 'cutoff_reason': cutoff['reason'],
+ 'native_rows': count, 'native_sha256': digest_native.hexdigest(),
+ 'commands': len(actions), 'commands_sha256': digest(actions)}
+
+
+def inspect(output):
+ from .piper_grip_benchmark import decode_attempt
+ from .probe_evidence import _arrays
+ output = Path(output)
+ protocol_path = output/'protocol.json'
+ if protocol_path.is_symlink() or not protocol_path.is_file() or protocol_path.stat().st_size > 16*1024*1024:
+ raise ValueError('Expected a bounded plain frozen protocol.')
+ with protocol_path.open('rb') as stream:
+ raw_protocol = stream.read(16*1024*1024+1)
+ if (len(raw_protocol) > 16*1024*1024
+ or sha256(raw_protocol).hexdigest() != read(output/'frozen.json')['protocol_sha256']):
+ raise ValueError('Frozen protocol digest changed.')
+ protocol, report = json.loads(raw_protocol), read(output/'report.json')
+ if prior.sources() != protocol['source_sha256'] or report['source_sha256'] != protocol['source_sha256']:
+ raise ValueError('Inspect with the frozen implementation.')
+ planned = roster(protocol)
+ ids = [case['id']+'-'+variant for case, variant in planned]
+ if ([row['id'] for row in report['rows']] != ids or len(ids) != 36
+ or report['attempted'] != 36 or protocol['planned_attempts'] != 36):
+ raise ValueError('Attempt roster/denominator differs from frozen protocol.')
+ checked, prefixes = [], {}
+ cutoffs = {case['id']: _common_cutoff([row for row in report['rows'] if row['scene_id'] == case['id']])
+ for case in protocol['cases']}
+ for row, (case, variant) in zip(report['rows'], planned):
+ folder = output/row['id']
+ if set(row['files']) != {'started.json', 'episode.json.gz', 'native.npz', 'native.json', 'evaluation.json'}:
+ raise ValueError('Unexpected attempt artifact roster.')
+ for name, expected in row['files'].items():
+ path = folder/name
+ if path.is_symlink() or not path.is_file() or path.stat().st_size > MAX_FILE:
+ raise ValueError('Expected bounded plain evidence file.')
+ raw = path.read_bytes()
+ if {'bytes': len(raw), 'sha256': sha256(raw).hexdigest()} != expected:
+ raise ValueError('Frozen artifact digest changed.')
+ episode = json.loads(decode_attempt((folder/'episode.json.gz').read_bytes()))
+ if (episode['id'] != row['id'] or episode['variant'] != variant or episode['seed'] != case['seed']
+ or episode['config'] != case['config'] or episode['metrics'] != row['metrics'] or episode['valid'] != row['valid']
+ or encoded(_compact_policy(episode.get('policy'))) != encoded(row['policy'])):
+ raise ValueError('Episode identity/config/metrics disagree.')
+ arrays, metadata = _read_native(folder/'native.npz'), read(folder/'native.json')
+ reevaluated = evaluate(episode, arrays, metadata, protocol)
+ if encoded(reevaluated) != encoded(row['evaluation']) or encoded(read(folder/'evaluation.json')) != encoded(reevaluated):
+ raise ValueError('Recomputed endpoint/receipt differs.')
+ native_validated, reconstruction, reason = False, False, None
+ try:
+ _arrays(arrays, metadata)
+ native_validated = True
+ if episode['valid'] and episode['error'] is None:
+ _controls(episode, arrays)
+ reconstruction = True
+ prefixes.setdefault(case['id'], []).append((variant, _prefix(episode, arrays, cutoff=cutoffs[case['id']])))
+ except (ValueError, TypeError, KeyError) as exc:
+ reason = str(exc)
+ if episode['valid'] and episode['error'] is None:
+ raise ValueError('Valid attempt failed native readback: '+reason) from exc
+ checked.append({'id': row['id'], 'native_rows': len(arrays.get('t_seconds', [])),
+ 'native_validated': native_validated, 'controls_recomputed': reconstruction,
+ 'retained_refusal': not reconstruction, 'reason': reason, 'evaluation_recomputed': True})
+ prefix_checks = []
+ for case in protocol['cases']:
+ group = prefixes.get(case['id'], [])
+ if len(group) == 3:
+ base = group[0][1]
+ if any(encoded(value) != encoded(base) for _, value in group[1:]):
+ raise ValueError('Acquisition prefixes differ before any hold treatment.')
+ prefix_checks.append({'scene_id': case['id'], 'verified': True, **base,
+ 'treatment_boundaries': cutoffs[case['id']]['treatment_boundaries']})
+ else:
+ prefix_checks.append({'scene_id': case['id'], 'verified': False, 'reason': 'invalid_attempt_prevents_complete_triplet'})
+ if encoded(aggregates(report['rows'])) != encoded(report['aggregates']):
+ raise ValueError('Aggregate counts differ from retained rows.')
+ return {'schema': 'dvidia.piper-retention-inspection/v1', 'attempts': len(checked),
+ 'native_rows': sum(row['native_rows'] for row in checked), 'rows': checked,
+ 'acquisition_prefixes': prefix_checks, 'aggregates': report['aggregates'], 'physical_robot_ready': False,
+ 'scope': 'Read-only recorded consistency and causal native predicate reconstruction; hashes do not authenticate external data or hardware.'}
+
+
+def main(argv=None):
+ parser = argparse.ArgumentParser(description=__doc__)
+ sub = parser.add_subparsers(dest='command', required=True)
+ for name in ('prepare', 'run'):
+ command = sub.add_parser(name)
+ command.add_argument('--output', type=Path, required=True)
+ command.add_argument('--assets', type=Path, required=True)
+ command = sub.add_parser('inspect'); command.add_argument('--output', type=Path, required=True)
+ args = parser.parse_args(argv)
+ try:
+ result = inspect(args.output) if args.command == 'inspect' else globals()[args.command](args.output, args.assets)
+ print(json.dumps(result if args.command == 'inspect' else {'schema': SCHEMA, 'planned_attempts': 36,
+ 'physical_robot_ready': False}, indent=2))
+ return 0
+ except (ValueError, OSError, KeyError, ImportError) as exc:
+ print('Retention study refused: '+str(exc), file=sys.stderr)
+ return 2
+
+
+if __name__ == '__main__':
+ raise SystemExit(main())
diff --git a/tests/test_piper_retention.py b/tests/test_piper_retention.py
new file mode 100644
index 0000000..237caa2
--- /dev/null
+++ b/tests/test_piper_retention.py
@@ -0,0 +1,231 @@
+"""Synthetic retention-controller boundaries; no native physics or hardware."""
+from copy import deepcopy
+import math
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+from dvidia_training import piper_retention as hold
+from dvidia_training import piper_task
+
+
+class Environment:
+ def __init__(self):
+ self.config = SimpleNamespace(control_dt=.02, timeout_s=20., tool_orientation_wxyz=[1.,0.,0.,0.],
+ goal_position_tolerance_m=.025)
+ self.model = SimpleNamespace(actuator_ctrlrange=[[-2.,2.]]*6+[[0.,.035]])
+ self.requests, self.raw = [], None
+ self.observation = None
+
+ def pose_action(self, position, quaternion, jaw):
+ self.requests.append((list(position),list(quaternion),jaw))
+ # A nonzero moving-goal servo offset becomes zero only at the exact
+ # admission pose, making issued and measured holds meaningfully differ.
+ offset = .03 if math.dist(position,self.observation['tcp_position_m']) > 1e-10 else 0.
+ return {'joint_targets_rad': list(self.raw) if self.raw is not None else [q+offset for q in self.observation['joint_position_rad']],
+ 'gripper_position_m': .012}
+
+
+class Teacher:
+ def __init__(self, env):
+ self.env, self.phase, self.issued_phase, self.ticks = env, 'close', 'close', 35
+ self.held_offset, self.grasp_position, self.calls = None, [.3,0.,.082], []
+
+ def __call__(self, observation):
+ if self.phase not in ('approach','descend','close'):
+ raise AssertionError('Full transport must not be dispatched')
+ self.calls.append(self.phase)
+ self.issued_phase = self.phase
+ self.phase, self.ticks = 'lift', 0
+ return {'joint_targets_rad': [.05]*6, 'gripper_position_m': .012}
+
+
+def observation(t, *, rise=None, force=.4, supported=False, both=True, slip=0., joints=.02, tcp_offset=0.):
+ rise = max(0.,min(.01,.02*(t-.02))) if rise is None else rise
+ return {'t_seconds':t,'joint_position_rad':[joints]*6,'gripper_joint_position_m':[.02,-.02],
+ 'tcp_position_m':[.3,0.,.082+rise+tcp_offset],'tcp_quaternion_wxyz':[1.,0.,0.,0.],
+ 'object_position_m':[.3+slip,0.,.08+rise+tcp_offset], 'target_position_m':[.35,.08,.08],
+ 'grasp_contact_counts':{'left':int(both),'right':int(both)},
+ 'pad_normal_forces_n':{'left':force,'right':force},'table_object_contact_count':int(supported),
+ 'unexpected_contact_count':0,'native_actuator_controls':[0.]*6+[.027]}
+
+
+def make(variant):
+ env = Environment()
+ with patch.object(piper_task,'PiperTaskPolicy',Teacher):value=hold.RetentionPolicy(env,variant)
+ return value,env
+
+
+def call(value,env,obs):
+ env.observation = obs
+ return value(obs)
+
+
+def admitted(value,env):
+ for index in range(40):
+ action=call(value,env,observation(index*.02))
+ if value.admitted:return action
+ raise AssertionError('Bounded synthetic fixture did not admit')
+
+
+class RetentionTests(unittest.TestCase):
+ def test_contract_has_fixed_pretrial_limits_and_explicit_scope(self):
+ contract=hold.retention_contract()
+ self.assertEqual(contract['variants'],['measured_joint','issued_joint','cartesian'])
+ self.assertEqual(contract['limits']['joint_correction_ceiling_rad'],.03)
+ self.assertEqual(contract['limits']['joint_correction_rate_rad_s'],.10)
+ self.assertFalse(contract['physical_robot_ready'])
+ for variant in ('evidence',True,None):
+ with self.assertRaises(ValueError):make(variant)
+
+ def test_prefix_is_exactly_equal_until_admission_and_no_full_teacher_dispatch(self):
+ values=[make(variant) for variant in hold.VARIANTS]
+ for index in range(40):
+ results=[call(value,env,observation(index*.02)) for value,env in values]
+ if any(value.admitted for value,_ in values):
+ self.assertTrue(all(value.admitted for value,_ in values));break
+ self.assertEqual(results[0],results[1]);self.assertEqual(results[0],results[2])
+ self.assertTrue(all(value.teacher.calls==['close'] for value,_ in values))
+ for value,_ in values:
+ self.assertEqual(value.admission_time_s,values[0][0].admission_time_s)
+
+ def test_measured_and_issued_freeze_distinct_actual_targets_without_jaw_change(self):
+ for variant in ('measured_joint','issued_joint'):
+ value,env=make(variant)
+ action=admitted(value,env)
+ expected=([.02]*6 if variant=='measured_joint' else value.hold_reference['issued_joint_targets_rad'])
+ self.assertEqual(action['joint_targets_rad'],expected)
+ self.assertEqual(action['gripper_position_m'],.012)
+ later=call(value,env,observation(value.last_observed_s+.02,joints=.04))
+ self.assertEqual(later,action)
+ if variant=='issued_joint':
+ self.assertNotEqual(action['joint_targets_rad'],[.02]*6)
+
+ def test_cartesian_admission_is_bumpless_then_correction_has_fixed_bias_and_rate(self):
+ value,env=make('cartesian');action=admitted(value,env)
+ reference=value.hold_reference
+ self.assertEqual(action['joint_targets_rad'],reference['issued_joint_targets_rad'])
+ self.assertEqual(action['gripper_position_m'],.012)
+ env.raw=[.03]*6
+ corrected=call(value,env,observation(value.last_observed_s+.02))
+ self.assertEqual(corrected['joint_targets_rad'],[baseline+.002 for baseline in reference['issued_joint_targets_rad']])
+ self.assertEqual(corrected['gripper_position_m'],.012)
+ record=value.hold_records[-1]
+ self.assertEqual(record['correction_rate_limited_axes'],list(range(6)))
+ self.assertEqual(record['correction_rad'],[.002]*6)
+ self.assertEqual(record['reference_position_m'],reference['position_m'])
+ self.assertEqual(record['fixed_bias_rad'],reference['fixed_bias_rad'])
+
+ def test_original_loss_and_force_ceiling_terminally_refuse_without_requalification(self):
+ for mutation,reason in (({'both':False},'bilateral_contact_or_force_loss'),
+ ({'supported':True},'support_returned'),
+ ({'force':10.0001},'declared_force_ceiling_exceeded'),
+ ({'slip':.0031},'relative_slip')):
+ for variant in hold.VARIANTS:
+ value,env=make(variant);admitted(value,env)
+ action=call(value,env,observation(value.last_observed_s+.02,joints=.04,**mutation))
+ self.assertFalse(value.admitted)
+ self.assertEqual(value.abort_reason,reason)
+ self.assertEqual(action,{'joint_targets_rad':[.04]*6,'gripper_position_m':.012})
+ again=call(value,env,observation(value.last_observed_s+.02))
+ self.assertEqual(again,action)
+ self.assertFalse(value.admitted)
+
+ def test_force_ceiling_also_stops_acquisition_before_teacher(self):
+ value,env=make('cartesian')
+ action=call(value,env,observation(0.,force=10.001))
+ self.assertEqual(value.teacher.calls,[])
+ self.assertEqual(action,{'joint_targets_rad':[.02]*6,'gripper_position_m':.027})
+ self.assertEqual(value.abort_reason,'declared_force_ceiling_exceeded')
+
+ def test_terminal_records_do_not_relabel_old_ik_or_rate_flags_as_current(self):
+ value,env=make('cartesian');admitted(value,env)
+ env.raw=[.03]*6
+ call(value,env,observation(value.last_observed_s+.02))
+ self.assertTrue(value.hold_records[-1]['correction_rate_limited'])
+ call(value,env,observation(value.last_observed_s+.02,force=10.001))
+ for record in (value.hold_records[-1],):
+ self.assertIsNone(record['raw_pose_joint_targets_rad'])
+ self.assertIsNone(record['desired_correction_rad'])
+ self.assertFalse(record['correction_rate_limited'])
+ call(value,env,observation(value.last_observed_s+.02))
+ self.assertIsNone(value.hold_records[-1]['raw_pose_joint_targets_rad'])
+ self.assertFalse(value.hold_records[-1]['correction_rate_limited'])
+
+ def test_position_orientation_correction_and_initial_bias_exhaustion_refuse(self):
+ for bound in ('position','orientation','correction','initial_bias'):
+ value,env=make('cartesian')
+ action=admitted(value,env) if bound!='initial_bias' else None
+ if bound=='initial_bias':
+ for index in range(40):
+ if value.qualification_since is not None and index*.02-value.qualification_since >= .1-1e-9:
+ env.raw=[-.2]*6
+ action=call(value,env,observation(index*.02))
+ if value.aborted:break
+ expected='initial_bias_exhausted'
+ else:
+ obs=observation(value.last_observed_s+.02)
+ if bound=='position':
+ obs=observation(obs['t_seconds'],tcp_offset=.0041)
+ expected='tcp_position_error_exhausted'
+ elif bound=='orientation':
+ angle=.051
+ obs['tcp_quaternion_wxyz']=[math.cos(angle/2),math.sin(angle/2),0.,0.]
+ expected='tcp_orientation_error_exhausted'
+ else:
+ env.raw=[.051]*6
+ expected='joint_correction_exhausted'
+ action=call(value,env,obs)
+ with self.subTest(bound=bound):
+ self.assertFalse(value.admitted)
+ self.assertEqual(value.abort_reason,expected)
+ self.assertEqual(action['joint_targets_rad'],[.02]*6)
+ self.assertEqual(action['gripper_position_m'],.012)
+
+ def test_native_margin_exhaustion_refuses_instead_of_clipping(self):
+ value,env=make('cartesian');admitted(value,env)
+ value.joint_limits=[[.001,.041]]*6
+ env.raw=[.022]*6
+ action=call(value,env,observation(value.last_observed_s+.02))
+ self.assertEqual(value.abort_reason,'native_joint_margin_exhausted')
+ self.assertFalse(value.admitted)
+ self.assertEqual(action['joint_targets_rad'],[.02]*6)
+
+ def test_invalid_measurement_refuses_without_new_command_or_untrusted_hold(self):
+ for bad in ({'joint_position_rad':[float('nan')]*6},{'t_seconds':.01}):
+ value,env=make('cartesian');admitted(value,env)
+ obs=observation(value.last_observed_s+.02);obs.update(bad)
+ before=len(value.requested_controls)
+ with self.assertRaises(ValueError):call(value,env,obs)
+ self.assertEqual(len(value.requested_controls),before)
+ self.assertFalse(value.admitted)
+ self.assertEqual(value.status,'invalid_measurement_refused')
+ self.assertFalse(value.input_refusals[-1]['command_issued'])
+ with self.assertRaises(ValueError):call(value,env,observation(value.last_observed_s+.02))
+
+ def test_final_observation_checks_force_and_error_without_another_solve_or_command(self):
+ for mutation,reason in (({'force':10.001},'declared_force_ceiling_exceeded'),
+ ({'tcp_offset':.0041},'tcp_position_error_exhausted'),
+ ({},None)):
+ value,env=make('cartesian');admitted(value,env)
+ requests,commands=len(env.requests),len(value.requested_controls)
+ result=value.observe_final(observation(value.last_observed_s+.02,**mutation))
+ self.assertEqual(len(env.requests),requests)
+ self.assertEqual(len(value.requested_controls),commands)
+ self.assertEqual(result['abort_reason'],reason)
+ self.assertEqual(result['admitted'],reason is None)
+ self.assertTrue(result['retention']['hold_records'][-1]['final_observation'])
+ self.assertFalse(result['retention']['hold_records'][-1]['command_issued'])
+
+ def test_summary_records_are_detached(self):
+ value,env=make('cartesian');admitted(value,env)
+ summary=value.summary()
+ self.assertEqual(summary['variant'],'cartesian')
+ self.assertEqual(summary['acquisition_variant'],'evidence')
+ summary['retention']['hold_reference']['position_m'][0]=99.
+ summary['retention']['hold_records'][0]['joint_targets_rad'][0]=99.
+ self.assertNotEqual(value.hold_reference['position_m'][0],99.)
+ self.assertNotEqual(value.hold_records[0]['joint_targets_rad'][0],99.)
+
+
+if __name__=='__main__':unittest.main()
diff --git a/tests/test_piper_retention_benchmark.py b/tests/test_piper_retention_benchmark.py
new file mode 100644
index 0000000..b179f1f
--- /dev/null
+++ b/tests/test_piper_retention_benchmark.py
@@ -0,0 +1,333 @@
+"""Fixed-protocol and native-shaped falsification tests; no physics runs."""
+from copy import deepcopy
+from hashlib import sha256
+from pathlib import Path
+from tempfile import TemporaryDirectory
+import unittest
+from unittest.mock import patch
+
+import numpy as np
+
+from dvidia_training import piper_retention_benchmark as study
+from dvidia_training import probe_evidence as proof
+
+
+def fixture(admission_s=2.):
+ count = 20000
+ arrays = {name: np.zeros((count,) if width == 1 else (count, width), dtype=np.float64)
+ for name, width in proof.FIELDS}
+ arrays['native_tick'][:] = np.arange(1, count+1)
+ arrays['t_seconds'][:] = arrays['native_tick']*.001
+ arrays['object_position_m'][:] = [.303, -.026, .082]
+ arrays['tcp_position_m'][:] = [.303, -.026, .182]
+ lifted = arrays['t_seconds'] >= 1.1
+ arrays['object_position_m'][lifted, 2] += .01
+ arrays['tcp_position_m'][lifted, 2] += .01
+ arrays['object_minus_tcp_local_m'][:] = arrays['object_position_m']-arrays['tcp_position_m']
+ arrays['object_quaternion_wxyz'][:, 0] = 1.
+ arrays['tcp_quaternion_wxyz'][:, 0] = 1.
+ arrays['grasp_contact_counts'][:] = [1., 1.]
+ arrays['pad_normal_forces_n'][:] = [.2, .3]
+ arrays['table_object_contact_count'][:] = ~lifted
+ metadata = {'schema': 'dvidia.piper-feedback-native/v1', 'native_timestep_s': .001,
+ 'rows': count, 'fields': [{'name': name, 'width': width} for name, width in proof.FIELDS],
+ 'side_order': ['left', 'right'], 'physical_robot_ready': False}
+ case = study.case_definitions()[0]
+ probe = {'probe_index': 0, 'start_time_s': 1.,
+ 'object_origin_m': arrays['object_position_m'][999].tolist(),
+ 'tcp_origin_m': arrays['tcp_position_m'][999].tolist(),
+ 'tcp_quaternion_wxyz': [1., 0., 0., 0.],
+ 'local_offset_origin_m': arrays['object_minus_tcp_local_m'][999].tolist(),
+ 'end_time_s': 20., 'disposition': 'admitted', 'reason': 'qualified'}
+ checks = [{'t_seconds': t, 'qualified': True, 'probe_index': 0}
+ for t in sorted({admission_s, min(20., admission_s+5.), 20.})]
+ result = {'id': case['id']+'-measured_joint', 'scene_id': case['id'], 'seed': case['seed'],
+ 'variant': 'measured_joint', 'config': case['config'], 'valid': True, 'error': None,
+ 'initial_object_position_m': arrays['object_position_m'][0].tolist(),
+ 'compiled_model_sha256': 'a'*64,
+ 'actions': [{} for _ in range(1000)], 'trajectory': [{} for _ in range(1001)],
+ 'metrics': {'valid': True, 'warnings': [], 'unexpected_contact_native_ticks': 0,
+ 'simulation_time': 20.},
+ 'policy': {'probes': [probe], 'admitted': True,
+ 'events': [{'type': 'evidence_qualified', 't_seconds': admission_s, 'probe_index': 0}],
+ 'qualification_checks': checks, 'requested_controls': [],
+ 'retention': {'hold_records': [], 'input_refusals': []}}}
+ protocol = {'force_ceiling_n': 10., 'source_sha256': {}, 'asset_lock_sha256': 'a'*64}
+ return result, arrays, metadata, protocol
+
+
+def endpoint(result, arrays, metadata, protocol):
+ # The bound receipt has its own integration test below. Endpoint mutations
+ # stay isolated from that existing module's terminal qualification semantics.
+ with patch.object(study.prior, '_evaluate', return_value={'verification': {'admissible': False}}):
+ return study.evaluate(result, arrays, metadata, protocol)
+
+
+def controls_fixture():
+ result, arrays, _, _ = fixture()
+ result['actions'], result['trajectory'] = [], []
+ result['policy']['requested_controls'] = []
+ control = np.zeros(7)
+ config = result['config']
+ maximum = np.array([config['joint_target_rate_rad_s']*.001]*6+[config['jaw_target_rate_m_s']*.001])
+ for i in range(1001):
+ sample = max(0, i*20-1)
+ obs = {field: arrays[field][sample].tolist()
+ for field in ('object_position_m', 'joint_position_rad', 'tcp_position_m')}
+ obs.update(native_actuator_controls=control.tolist(), command=None,
+ issued_joint_targets_rad=None, issued_gripper_position_m=None)
+ result['trajectory'].append(obs)
+ if i == 1000:
+ break
+ action = {'t_seconds': i*.02, 'joint_targets_rad': [(.01 if i % 2 else -.01), 0., 0., 0., 0., 0.],
+ 'gripper_position_m': .005}
+ result['actions'].append(action)
+ result['policy']['requested_controls'].append({'index': i, **deepcopy(action)})
+ obs.update(command=deepcopy(action), issued_joint_targets_rad=list(action['joint_targets_rad']),
+ issued_gripper_position_m=action['gripper_position_m'])
+ target = np.array(action['joint_targets_rad']+[action['gripper_position_m']])
+ for tick in range(i*20, (i+1)*20):
+ control += np.clip(target-control, -maximum, maximum)
+ arrays['arm_actuator_controls_rad'][tick] = control[:6]
+ arrays['applied_jaw_target_m'][tick] = control[-1]
+ arrays['requested_jaw_target_m'][tick] = target[-1]
+ return result, arrays
+
+
+class RetentionStudyTests(unittest.TestCase):
+ def test_fresh_fixed_cells_balanced_order_and_every_attempt_once(self):
+ cases = study.case_definitions()
+ self.assertEqual(len(cases), 12)
+ self.assertEqual([case['seed'] for case in cases], list(range(311100, 311112)))
+ self.assertEqual({case['config']['mass_kg'] for case in cases}, {.045, .055})
+ self.assertEqual({case['config']['friction'] for case in cases}, {.75, .50, .20})
+ self.assertEqual({tuple(case['config']['object_position_m'][:2]) for case in cases},
+ {(.303, -.026), (.307, -.037)})
+ roster = study.roster({'cases': cases})
+ self.assertEqual(len({(case['id'], variant) for case, variant in roster}), 36)
+ for variant in study.VARIANTS:
+ self.assertEqual([sum(case['variant_order'][position] == variant for case in cases)
+ for position in range(3)], [4, 4, 4])
+
+ def test_correct_five_second_retention_and_bound_receipt_integration(self):
+ result, arrays, metadata, protocol = fixture()
+ evaluation = study.evaluate(result, arrays, metadata, protocol)
+ self.assertTrue(evaluation['primary']['success'])
+ self.assertTrue(evaluation['stationarity']['grip_retention_and_stationarity'])
+ self.assertTrue(evaluation['receipt']['verification']['admissible'])
+ self.assertFalse(evaluation['physical_robot_ready'])
+
+ def test_one_ms_force_gap_and_ceiling_breach_reset_admission_prefix(self):
+ for force in (.01, 10.000001):
+ result, arrays, metadata, protocol = fixture()
+ arrays['pad_normal_forces_n'][1949, 0] = force
+ evaluation = endpoint(result, arrays, metadata, protocol)
+ self.assertFalse(evaluation['first_admission']['native_prefix_qualified'])
+ self.assertEqual(evaluation['primary']['reason'], 'false_initial_admission')
+ self.assertFalse(evaluation['primary']['success'])
+
+ def test_99_native_prefix_samples_do_not_equal_100(self):
+ result, arrays, _, _ = fixture()
+ probe = result['policy']['probes'][0]
+ _, streak, qualified, _, _ = study.probe_series(arrays, probe)
+ self.assertEqual(streak[1197], 99)
+ self.assertFalse(qualified[1197])
+ self.assertEqual(streak[1198], 100)
+ self.assertTrue(qualified[1198])
+
+ def test_five_second_endpoint_includes_all_5000_followup_ticks(self):
+ for failure_time, success in ((6.999, False), (7., False), (7.001, True)):
+ result, arrays, metadata, protocol = fixture()
+ arrays['pad_normal_forces_n'][round(failure_time/.001)-1, 0] = 0.
+ evaluation = endpoint(result, arrays, metadata, protocol)
+ self.assertEqual(evaluation['primary']['success'], success, failure_time)
+ self.assertEqual(evaluation['primary']['complete_5s'], True)
+
+ def test_less_than_five_seconds_followup_is_never_padded(self):
+ for admission, complete in ((15., True), (15.02, False)):
+ result, arrays, metadata, protocol = fixture(admission)
+ evaluation = endpoint(result, arrays, metadata, protocol)
+ self.assertEqual(evaluation['primary']['success'], complete)
+ self.assertEqual(evaluation['primary']['complete_5s'], complete)
+ if not complete:
+ self.assertEqual(evaluation['primary']['reason'], 'insufficient_followup')
+
+ def test_native_retention_does_not_resurrect_controller_abstention(self):
+ result, arrays, metadata, protocol = fixture()
+ result['policy']['admitted'] = False
+ for claim in result['policy']['qualification_checks'][1:]:
+ claim['qualified'] = False
+ evaluation = study.evaluate(result, arrays, metadata, protocol)
+ self.assertTrue(evaluation['primary']['success'])
+ self.assertFalse(evaluation['primary']['controller_retained_at_5s'])
+ self.assertFalse(evaluation['primary']['combined_qualified_pause'])
+ self.assertFalse(evaluation['final']['both_qualified'])
+ self.assertFalse(evaluation['final']['evidence_admissible'])
+
+ def test_no_admission_and_invalid_native_attempts_are_retained_as_failures(self):
+ result, arrays, metadata, protocol = fixture()
+ result['policy']['events'] = []
+ result['policy']['admitted'] = False
+ evaluation = endpoint(result, arrays, metadata, protocol)
+ self.assertEqual(evaluation['primary']['reason'], 'no_admission')
+ self.assertIsNone(evaluation['first_admission'])
+ self.assertFalse(evaluation['stationarity']['grip_retention_and_stationarity'])
+ for field in ('warning', 'incomplete', 'critical_count_mismatch', 'critical_before_probe', 'malformed_sidecar'):
+ changed, data, meta, protocol = fixture()
+ if field == 'warning': changed['metrics']['warnings'] = ['native warning']
+ elif field == 'incomplete': changed['actions'].pop()
+ elif field == 'critical_count_mismatch': data['unexpected_contact_count'][100] = 1.
+ elif field == 'critical_before_probe':
+ data['unexpected_contact_count'][100] = 1.
+ changed['metrics']['unexpected_contact_native_ticks'] = 1
+ else: data['native_tick'][100] += 1.
+ evaluation = endpoint(changed, data, meta, protocol)
+ self.assertFalse(evaluation['primary']['success'], field)
+ self.assertEqual(evaluation['status'], 'refused')
+
+ def test_broader_post_endpoint_force_ceiling_veto_is_preserved(self):
+ result, arrays, metadata, protocol = fixture()
+ arrays['pad_normal_forces_n'][15000, 0] = 10.0001
+ evaluation = endpoint(result, arrays, metadata, protocol)
+ self.assertFalse(evaluation['primary']['success'])
+ self.assertEqual(evaluation['primary']['reason'], 'force_ceiling_exceeded')
+ self.assertFalse(evaluation['primary']['all_native_hold_forces_within_ceiling'])
+
+ def test_common_tcp_position_and_orientation_endpoint_for_all_variants(self):
+ for variant in study.VARIANTS:
+ result, arrays, metadata, protocol = fixture()
+ result['variant'] = variant
+ # The tool and object move together: retained grip alone cannot
+ # qualify a stationary pause.
+ arrays['tcp_position_m'][2999:, 0] += .004001
+ arrays['object_position_m'][2999:, 0] += .004001
+ evaluation = endpoint(result, arrays, metadata, protocol)
+ self.assertTrue(evaluation['primary']['success'])
+ self.assertFalse(evaluation['stationarity']['grip_retention_and_stationarity'])
+ self.assertAlmostEqual(evaluation['stationarity']['maximum_tcp_position_excursion_m'], .004001)
+ result, arrays, metadata, protocol = fixture()
+ angle = .050001
+ arrays['tcp_quaternion_wxyz'][2999:] = [np.cos(angle/2), 0., 0., np.sin(angle/2)]
+ evaluation = endpoint(result, arrays, metadata, protocol)
+ self.assertTrue(evaluation['primary']['success'])
+ self.assertFalse(evaluation['stationarity']['grip_retention_and_stationarity'])
+ self.assertAlmostEqual(evaluation['stationarity']['maximum_tcp_orientation_excursion_rad'], angle)
+ arrays['tcp_quaternion_wxyz'][2999:] *= -1 # Quaternion sign changes no pose.
+ opposite = endpoint(result, arrays, metadata, protocol)
+ self.assertEqual(evaluation['stationarity'], opposite['stationarity'])
+
+ def test_exact_stationarity_bounds_pass_and_terminal_recovery_cannot_hide_excursion(self):
+ result, arrays, metadata, protocol = fixture()
+ arrays['tcp_position_m'][2999:3999, 0] += .004
+ arrays['object_position_m'][2999:3999, 0] += .004
+ angle = .05
+ arrays['tcp_quaternion_wxyz'][2999:3999] = [np.cos(angle/2), 0., 0., np.sin(angle/2)]
+ # With a z-only local offset, yaw preserves the redundant local field.
+ evaluation = endpoint(result, arrays, metadata, protocol)
+ self.assertTrue(evaluation['stationarity']['grip_retention_and_stationarity'])
+ self.assertEqual(evaluation['stationarity']['terminal_tcp_position_delta_m'], [0., 0., 0.])
+ arrays['tcp_position_m'][3499, 0] += .000001
+ arrays['object_position_m'][3499, 0] += .000001
+ changed = endpoint(result, arrays, metadata, protocol)
+ self.assertFalse(changed['stationarity']['grip_retention_and_stationarity'])
+ self.assertEqual(changed['stationarity']['terminal_tcp_position_delta_m'], [0., 0., 0.])
+
+ def test_corrected_control_tape_requires_every_native_slew_sample_and_boundary(self):
+ result, arrays = controls_fixture()
+ study._controls(result, arrays)
+ arrays['arm_actuator_controls_rad'][54, 0] += 1e-6
+ with self.assertRaisesRegex(ValueError, 'slew law'):
+ study._controls(result, arrays)
+ result, arrays = controls_fixture()
+ result['trajectory'][4]['issued_gripper_position_m'] += .001
+ with self.assertRaisesRegex(ValueError, 'disagree'):
+ study._controls(result, arrays)
+
+ def test_prefix_excludes_first_hold_command_but_includes_admission_native_row(self):
+ result, arrays = controls_fixture()
+ original = study._prefix(result, arrays)
+ result['actions'][100]['joint_targets_rad'][0] += .01 # Command at admission=2s.
+ arrays['joint_position_rad'][2000, 0] += .01 # First post-admission native tick.
+ self.assertEqual(study._prefix(result, arrays), original)
+ arrays['joint_position_rad'][1999, 0] += .01
+ self.assertNotEqual(study._prefix(result, arrays)['native_sha256'], original['native_sha256'])
+
+ def test_cartesian_pre_admission_bound_refusal_keeps_common_acquisition_proof(self):
+ measured, arrays = controls_fixture()
+ issued = deepcopy(measured); issued['variant'] = 'issued_joint'
+ cartesian = deepcopy(measured); cartesian['variant'] = 'cartesian'
+ cartesian['policy']['events'] = [{'type': 'hold_invalidated', 't_seconds': 2.,
+ 'reason': 'initial_bias_exhausted', 'probe_index': 0}]
+ cutoff = study._common_cutoff([measured, issued, cartesian])
+ self.assertEqual(cutoff['time_s'], 2.)
+ self.assertIn('pre_admission_bound_refusal', cutoff['reason'])
+ self.assertEqual(len(cutoff['treatment_boundaries']), 3)
+ # Different treatment command at the common boundary is expected.
+ cartesian['actions'][100]['joint_targets_rad'][0] += .01
+ prefixes = [study._prefix(row, arrays, cutoff=cutoff) for row in (measured, issued, cartesian)]
+ self.assertEqual(prefixes[0], prefixes[1]); self.assertEqual(prefixes[0], prefixes[2])
+ self.assertEqual(prefixes[0]['native_rows'], 2000)
+ self.assertEqual(prefixes[0]['commands'], 100)
+ cartesian['actions'][99]['joint_targets_rad'][0] += .01
+ self.assertNotEqual(study._prefix(cartesian, arrays, cutoff=cutoff)['commands_sha256'],
+ prefixes[0]['commands_sha256'])
+
+ def test_acquisition_alarm_is_not_a_treatment_cutoff_and_no_treatment_uses_full_run(self):
+ result, arrays = controls_fixture()
+ result['policy']['events'].insert(0, {'type': 'alarm', 't_seconds': 1.5, 'reason': 'relative_slip'})
+ self.assertEqual(study._common_cutoff([result])['time_s'], 2.)
+ result['policy']['events'] = [{'type': 'hold_invalidated', 't_seconds': 1.5,
+ 'reason': 'bilateral_contact_or_force_loss'}]
+ cutoff = study._common_cutoff([result])
+ self.assertIsNone(cutoff['time_s'])
+ prefix = study._prefix(result, arrays, cutoff=cutoff)
+ self.assertEqual(prefix['native_rows'], 20000)
+ self.assertEqual(prefix['commands'], 1000)
+
+ def test_correction_summary_does_not_mix_terminal_checks_with_issued_controls(self):
+ record = {'command_issued': True, 'correction_rate_limited': True,
+ 'correction_rate_limited_axes': [0, 2], 'correction_rad': [.01]*6,
+ 'desired_correction_rad': [.02]*6, 'refused_bounds': None}
+ final = {**record, 'command_issued': False, 'refused_bounds': 'tcp_position_error_exhausted'}
+ summary = study._corrections({'retention': {'hold_records': [record, final], 'input_refusals': []}})
+ self.assertEqual(summary['hold_record_count'], 2)
+ self.assertEqual(summary['bound_refusal_counts'], {'tcp_position_error_exhausted': 1})
+ self.assertEqual(summary['rate_limited_axis_occurrences'], 4)
+ self.assertEqual(summary['maximum_abs_active_correction_rad'], .01)
+ self.assertEqual(summary['maximum_abs_desired_correction_rad'], .02)
+
+ def test_existing_output_and_started_study_are_not_overwritten(self):
+ with TemporaryDirectory() as folder, patch.object(study, 'make_protocol') as make:
+ with self.assertRaises(ValueError): study.prepare(folder, 'unused-assets')
+ make.assert_not_called()
+ protocol = {'schema': study.SCHEMA, 'cases': []}
+ with TemporaryDirectory() as folder, patch.object(study, 'make_protocol', return_value=protocol), \
+ patch.object(study, '_attempt') as attempt:
+ study.write(Path(folder)/'protocol.json', protocol)
+ study.write(Path(folder)/'started.json', {'original': 'retained'})
+ with self.assertRaises(FileExistsError): study.run(folder, 'unused-assets')
+ attempt.assert_not_called()
+ self.assertEqual(study.read(Path(folder)/'started.json'), {'original': 'retained'})
+
+ def test_changed_protocol_refuses_before_native_attempt(self):
+ original = {'schema': study.SCHEMA, 'planned_attempts': 36, 'cases': []}
+ with TemporaryDirectory() as folder, patch.object(study, 'make_protocol', return_value=original), \
+ patch.object(study, '_attempt') as attempt:
+ study.write(Path(folder)/'protocol.json', {**original, 'planned_attempts': 35})
+ with self.assertRaises(ValueError): study.run(folder, 'unused-assets')
+ attempt.assert_not_called()
+ self.assertFalse((Path(folder)/'started.json').exists())
+
+ def test_inspection_checks_frozen_protocol_bytes_before_any_report_or_source_read(self):
+ with TemporaryDirectory() as folder, patch.object(study.prior, 'sources') as sources:
+ path = Path(folder)/'protocol.json'
+ study.write(path, {'planned_attempts': 36})
+ study.write(Path(folder)/'frozen.json', {'protocol_sha256': sha256(path.read_bytes()).hexdigest()})
+ path.write_bytes(path.read_bytes().replace(b'36', b'35'))
+ with self.assertRaisesRegex(ValueError, 'Frozen protocol digest changed'):
+ study.inspect(folder)
+ sources.assert_not_called()
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/tests/test_piper_retention_independent_audit.py b/tests/test_piper_retention_independent_audit.py
new file mode 100644
index 0000000..5e370d9
--- /dev/null
+++ b/tests/test_piper_retention_independent_audit.py
@@ -0,0 +1,152 @@
+"""Pure evidence audits; no simulation/runtime imports or native runs."""
+import sys
+from pathlib import Path
+import unittest
+
+import numpy as np
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'benchmarks' / 'piper-retention-v02'))
+import independent_audit as audit
+
+
+class TreatmentBoundaryTests(unittest.TestCase):
+ def test_earliest_pre_admission_bound_refusal_is_cutoff_but_later_refusal_is_not(self):
+ policy = {'events': [
+ {'type': 'hold_invalidated', 'reason': 'tcp_position_error_exhausted', 't_seconds': 1.02},
+ {'type': 'evidence_qualified', 'reason': 'sampled', 't_seconds': 2.},
+ {'type': 'hold_invalidated', 'reason': 'tcp_orientation_error_exhausted', 't_seconds': 3.},
+ {'type': 'hold_invalidated', 'reason': 'relative_slip', 't_seconds': .5},
+ ]}
+ self.assertEqual(audit.treatment_boundary(policy), {
+ 'time_s': 1.02, 'reason': 'pre_admission_bound_refusal',
+ 'detail': 'tcp_position_error_exhausted'})
+
+ def test_only_declared_bound_reasons_and_admissions_create_cutoffs(self):
+ self.assertIsNone(audit.treatment_boundary({'events': [
+ {'type': 'hold_invalidated', 'reason': 'relative_slip', 't_seconds': 1.},
+ {'type': 'abort', 'reason': 'pose_solver_refused', 't_seconds': 2.},
+ ]}))
+ self.assertEqual(audit.treatment_boundary({'events': [
+ {'type': 'evidence_qualified', 'reason': 'sampled', 't_seconds': 2.},
+ {'type': 'hold_invalidated', 'reason': 'joint_correction_exhausted', 't_seconds': 2.1},
+ ]})['reason'], 'admission')
+
+ def test_common_cutoff_combines_tied_reasons_deterministically(self):
+ result = audit.common_cutoff({
+ 'measured_joint': {'time_s': 1., 'reason': 'admission', 'detail': 'evidence_qualified'},
+ 'issued_joint': {'time_s': 1., 'reason': 'pre_admission_bound_refusal', 'detail': 'pose_solver_refused'},
+ 'cartesian': None,
+ })
+ self.assertEqual(result['time_s'], 1.)
+ self.assertEqual(result['reason'], 'admission+pre_admission_bound_refusal')
+ self.assertEqual([row['variant'] for row in result['treatment_boundaries']],
+ ['issued_joint', 'measured_joint'])
+
+ def test_invalid_triplet_remains_unknown_instead_of_asserting_three_way_match(self):
+ attempts = {variant: {'valid': True} for variant in audit.VARIANTS}
+ attempts['cartesian'] = {'valid': False, 'reason': 'invalid native sidecar',
+ 'native_rows_available': 100}
+ result = audit.paired_prefix('scene-1', attempts)
+ self.assertFalse(result['verified'])
+ self.assertEqual(result['reason'], 'invalid_attempt_prevents_complete_triplet')
+ self.assertEqual(result['variants'][2]['reason'], 'invalid native sidecar')
+
+
+class PrefixTests(unittest.TestCase):
+ def test_common_prefix_includes_cutoff_native_row_and_excludes_cutoff_command(self):
+ times = np.arange(1, 20001, dtype=np.float64)*.001
+ arrays = {'t_seconds': times, 'signal': np.arange(20000, dtype=np.float64)}
+ actions = [{'t_seconds': i*.02, 'command': i} for i in range(1000)]
+ cutoff = {'time_s': .04, 'reason': 'admission', 'treatment_boundaries': []}
+ result = audit.prefix_identity(arrays, actions, cutoff)
+ self.assertEqual(result['native_rows'], 40)
+ self.assertEqual(result['commands'], 2)
+ self.assertEqual(result['cutoff_time_s'], .04)
+ self.assertEqual(result['cutoff_reason'], 'admission')
+
+
+class StationarityTests(unittest.TestCase):
+ def arrays(self):
+ return {'tcp_position_m': np.zeros((4, 3), dtype=np.float64),
+ 'tcp_quaternion_wxyz': np.tile([1., 0., 0., 0.], (4, 1))}
+
+ def test_stationary_window_and_quaternion_sign_invariance(self):
+ arrays = self.arrays()
+ arrays['tcp_quaternion_wxyz'][2:] *= -1.
+ actual = audit.stationarity(arrays, 0, 3, True, True)
+ self.assertTrue(actual['tcp_pose_within_bounds'])
+ self.assertTrue(actual['grip_retention_and_stationarity'])
+ report = {'complete_5s': True, **actual}
+ audit.check_stationarity_reported(actual, report, True)
+
+ def test_position_orientation_and_incomplete_windows_fail_stationarity(self):
+ arrays = self.arrays(); arrays['tcp_position_m'][2, 0] = .004001
+ actual = audit.stationarity(arrays, 0, 3, True, True)
+ self.assertFalse(actual['tcp_pose_within_bounds'])
+ self.assertFalse(actual['grip_retention_and_stationarity'])
+ arrays = self.arrays(); theta = .050001
+ arrays['tcp_quaternion_wxyz'][2:] = [np.cos(theta/2), 0., 0., np.sin(theta/2)]
+ actual = audit.stationarity(arrays, 0, 3, True, True)
+ self.assertFalse(actual['tcp_pose_within_bounds'])
+ incomplete = audit.stationarity(self.arrays(), 0, 2, False, True)
+ self.assertFalse(incomplete['tcp_pose_within_bounds'])
+ self.assertFalse(incomplete['grip_retention_and_stationarity'])
+
+
+def correction_fixture():
+ issued = [.1]*6
+ raw0 = [.1]*6
+ jaw = .02
+ records, actions = [], []
+ for index, (time, delta, desired_delta, limited) in enumerate(
+ ((1., 0., 0., []), (1.02, .001, .001, []), (1.04, .003, .01, [0]))):
+ correction = [delta]+[0.]*5
+ desired = [desired_delta]+[0.]*5
+ raw = [raw0[0]+desired_delta]+raw0[1:]
+ targets = [base+offset for base, offset in zip(issued, correction)]
+ record = {'t_seconds': time, 'command_issued': True, 'fixed_bias_rad': [0.]*6,
+ 'joint_targets_rad': targets, 'frozen_jaw_target_m': jaw,
+ 'raw_pose_joint_targets_rad': raw, 'desired_correction_rad': desired,
+ 'correction_rad': correction, 'correction_rate_limited_axes': limited,
+ 'correction_rate_limited': bool(limited), 'refused_bounds': None}
+ records.append(record)
+ actions.append({'t_seconds': time, 'joint_targets_rad': targets,
+ 'gripper_position_m': jaw})
+ reference = {'t_seconds': 1., 'issued_joint_targets_rad': issued,
+ 'measured_joint_position_rad': issued, 'raw_pose_joint_targets_rad': raw0,
+ 'fixed_bias_rad': [0.]*6, 'frozen_jaw_target_m': jaw}
+ policy = {'variant': 'cartesian', 'frozen_jaw_target_m': jaw,
+ 'events': [{'type': 'evidence_qualified', 't_seconds': 1., 'reason': 'sampled'}],
+ 'retention': {'hold_reference': reference, 'hold_records': records, 'input_refusals': []}}
+ return policy, actions
+
+
+class CorrectionAuditTests(unittest.TestCase):
+ def test_cartesian_bias_first_command_correction_envelope_and_rate_are_checked(self):
+ policy, actions = correction_fixture()
+ result = audit.corrections_audit(policy, actions, audit.treatment_boundary(policy), 'cartesian')
+ self.assertEqual(result['hold_record_count'], 3)
+ self.assertEqual(result['rate_limited_axis_occurrences'], 1)
+ self.assertEqual(result['maximum_abs_active_correction_rad'], .003)
+ self.assertEqual(result['maximum_abs_desired_correction_rad'], .01)
+
+ def test_bad_first_target_bias_or_rate_exceedance_is_rejected(self):
+ policy, actions = correction_fixture()
+ policy['retention']['hold_records'][0]['joint_targets_rad'][0] += .001
+ actions[0]['joint_targets_rad'][0] += .001
+ with self.assertRaises(AssertionError):
+ audit.corrections_audit(policy, actions, audit.treatment_boundary(policy), 'cartesian')
+ policy, actions = correction_fixture()
+ policy['retention']['hold_reference']['fixed_bias_rad'][0] = .081
+ with self.assertRaises(AssertionError):
+ audit.corrections_audit(policy, actions, audit.treatment_boundary(policy), 'cartesian')
+ policy, actions = correction_fixture()
+ policy['retention']['hold_records'][2]['correction_rad'][0] = .004
+ policy['retention']['hold_records'][2]['joint_targets_rad'][0] = .104
+ actions[2]['joint_targets_rad'][0] = .104
+ with self.assertRaises(AssertionError):
+ audit.corrections_audit(policy, actions, audit.treatment_boundary(policy), 'cartesian')
+
+
+if __name__ == '__main__':
+ unittest.main()