Experiment formal custody reader red plan 01 - kmosoti/FabricO11y GitHub Wiki
Bounded custody reader controls before repair
PROSPECTIVE SOURCE-ONLY CONTROL CELL; no VM/custody/installation qualification. C061/X106 admits420 lines/max+120 production after297line3152;353 trust adds7literal methods before implementation, keeps all prior controls and oracles unchanged.
Clean harness 353455b8a90429357d9bc198152e19d0bc8fb3f0; driver SHA256 12dad7e1756c373dc73c04f76899b7cc6489bd357e263b091da4ae89ccebe309.
{
"tools/qualification/install/custody_witness.py": "3b23730f020463f2d05f9f1ba84ce07d1fe4dbdf7a6aed5d7e985cfb9d82fee1",
"tools/qualification/install/test_custody_witness.py": "27ad5a252bfacd51e1625db73c2bb8773c631dd52041a709d265509b8e696e9d",
"docs/experiments/formal/installation-release-protocol.md": "cb8dd0ba450423219b1f909b4fd9f552ba134c1bb88b56a70973ffd42956a5c0",
"tools/qualification/query_oracle.py": "fc696ee1173093c258ff4451c4a796dd8e5832761f9e0ff49f89a922b411d23a"
}
Exactly14 methods via actual file main. Default16/20GiB/swap0/1800s/full1GiB/fresh complete DATA namespace census95GB. No Cargo/native/guest. Expected old-reader red: valid matched groups/packed forms rejected; missed structural/schema/depth/container refusals; incorrect last-oneof/merged-body projection. Original three malformed nested cases unchanged. Opaque metric/trace nested leaves explicitly admitted only by this limited witness, not semantic product acceptance; originalpayload/SHA staysauthority. Tiny literal fixtures only, no childworkloads, source/staging unchanged. Preserve actual red, stop/post unexpected setup or missing positive; root re-review before any production repair.
python3 -B /run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-custody-reader-red-01.py installation-custody-reader-red-01 1073741824 python3 -B tools/qualification/install/test_custody_witness.py
tools/qualification/install/test_custody_witness.py
"""Prospective stopped-custody controls; fixture bytes never come from Rust."""
import copy
import hashlib
import importlib.util
import os
from pathlib import Path
import stat
import struct
import tempfile
import unittest
import zlib
ADAPTER = Path(__file__).with_name('custody_witness.py')
SOURCE = '/var/log/fabric-accept/allowed.log'
NODE = bytes(range(1, 17))
PENDING = b'accept-upgrade-pending-literal\n'
UNREAD = b'accept-upgrade-unread-literal\n'
BINDING = dict(run_id='custody-literal',
candidate_frozen='8fb6f413179b094288316e25845c137440ba23fa',
candidate_package_sha256='8edfc7709f4b5dc3da01c00a1d8c442470298e96f8c8e502cfb14e8c9e713e92',
package_source_commit='73ff77867d10ef521ca985b094e1d177f7786b77',
harness_commit='fcd57f62581a243fad34c61a258acb760cbdfce8')
PROCESS = dict(unit='fabrico11y-node.service', pid=55, boot_id='literal-boot-a',
start_ns=100, end_ns=200, **BINDING)
def varint(value):
out = bytearray()
while value >= 128:
out.append((value & 127) | 128); value >>= 7
return bytes(out + bytes([value]))
def vi(tag, value):
return varint(tag << 3) + varint(value)
def ld(tag, value):
return varint(tag << 3 | 2) + varint(len(value)) + value
def export_record(record, *, resource=b'', scope=b''):
return ld(1, (ld(1, resource) if resource else b'')
+ ld(2, (ld(1, scope) if scope else b'') + ld(2, record)))
def cursor(offset, *, prefix_crc=None, skipping=0, identity=b''):
prefix = (PENDING + UNREAD)[:min(offset, 64)]
return (ld(1, SOURCE.encode()) + vi(2, 7) + vi(3, 9) + vi(4, offset)
+ vi(5, skipping) + vi(6, len(prefix))
+ vi(7, zlib.crc32(prefix) if prefix_crc is None else prefix_crc)
+ (ld(8, identity) if identity else b''))
def batch(sequence=4, *, version=1, node=NODE, generation=1, body=PENDING,
offset=None, cursor_bytes=None, logs_bytes=None):
offset = len(PENDING) if offset is None else offset
attrs = {'log.file.path': SOURCE, 'log.file.device': '7', 'log.file.inode': '9',
'log.file.offset.start': str(offset - len(body)),
'log.file.offset.end': str(offset)}
record = b'\x59' + struct.pack('<Q', 123) + ld(5, ld(1, body[:-1]))
for key, value in attrs.items():
record += ld(6, ld(1, key.encode()) + ld(2, ld(1, value.encode())))
logs = ld(1, ld(2, ld(2, record))) if logs_bytes is None else logs_bytes
return (vi(1, version) + ld(2, node) + vi(3, generation) + vi(4, sequence)
+ ld(6, logs) + ld(7, cursor(offset) if cursor_bytes is None else cursor_bytes))
def frame(payload, at=0):
first = b'FAB1' + struct.pack('<I', len(payload))
marker = b'FAC1' + struct.pack('<Q', at + 16 + len(payload))
return (first + struct.pack('<II', zlib.crc32(first), zlib.crc32(payload))
+ payload + marker + struct.pack('<I', zlib.crc32(marker)))
def adapter():
spec = importlib.util.spec_from_file_location('custody_witness', ADAPTER)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
class MechanismAvailability(unittest.TestCase):
def test_stopped_snapshot_and_conservation_are_available(self):
self.assertTrue(ADAPTER.is_file(), 'NOT RUN: independent custody adapter is absent')
module = adapter()
self.assertTrue(callable(getattr(module, 'stopped_snapshot', None)))
self.assertTrue(callable(getattr(module, 'conserve', None)))
class CustodyWitnessControls(unittest.TestCase):
@classmethod
def setUpClass(cls):
if not ADAPTER.is_file():
raise unittest.SkipTest('NOT RUN: independent custody adapter is absent')
cls.module = adapter()
def setUp(self):
self.temp = tempfile.TemporaryDirectory(dir=os.environ['FABRIC_SCRATCH_ROOT'])
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
archive = self.root / 'allowed.log'; archive.write_bytes(PENDING + UNREAD)
self.logs = {SOURCE: dict(archive_path=str(archive), device=7, inode=9,
btrfs_identity=None)}
def snapshot(self, name, payloads=None, *, ack=3, change=None, receipt_change=None):
root = self.root / name; root.mkdir()
payloads = [batch()] if payloads is None else payloads
raw = b''
for payload in payloads: raw += frame(payload, len(raw))
(root / 'identity').write_bytes(b'FAI1' + NODE + struct.pack('<Q', 1))
(root / 'acked').write_bytes(b'FAK1' + struct.pack('<QQ', 1, ack))
(root / 'writer.lock').touch()
(root / 'batches.faj').write_bytes(raw)
receipt = dict(binding=dict(BINDING), spool_root=str(root), spool_bytes=268435456,
stop_confirmed=True,
boot_id='literal-boot-a', extra_project_units=[],
units={unit: dict(active_state='inactive', query_exit=3, main_pid=0,
cgroup_empty=True, companions=[])
for unit in ('fabrico11y-server.service', 'fabrico11y-node.service')},
candidate_process=copy.deepcopy(PROCESS) if name == 'drain' else None)
if change: change(root)
if receipt_change: receipt_change(receipt)
def inventory():
return {path.name: (path.lstat().st_mode,
str(path.readlink()) if path.is_symlink() else
path.read_bytes() if stat.S_ISREG(path.lstat().st_mode) else None)
for path in root.iterdir()}
original = inventory()
try:
return self.module.stopped_snapshot(root, receipt, self.logs)
finally:
self.assertEqual(inventory(), original, 'snapshot reader mutated raw source')
def chain(self):
before = self.snapshot('before')
installed = self.snapshot('installed')
drained = self.snapshot('drain', [batch(), batch(5, body=UNREAD,
offset=len(PENDING + UNREAD))], ack=5)
old = [dict(node='spindle-1', node_id=NODE.hex(), sequence=seq, index=0,
observed_ns=120, body=body, attributes={}) for seq, body in enumerate(
('accept-line-before-start', 'accept-line-during-outage',
'accept-line-after-pressure'), 1)]
rows = copy.deepcopy(old)
for sequence, body, start in ((4, PENDING, 0), (5, UNREAD, len(PENDING))):
rows.append(dict(node='spindle-1', node_id=NODE.hex(), sequence=sequence,
index=0, observed_ns=123, body=body[:-1].decode(), attributes={
'log.file.path': SOURCE, 'log.file.device': '7', 'log.file.inode': '9',
'log.file.offset.start': str(start), 'log.file.offset.end': str(start + len(body))}))
queries = dict(before=old, after_drain=copy.deepcopy(rows),
after_reboot=copy.deepcopy(rows), boot_before='literal-boot-a',
boot_after='literal-boot-b', pending_body=PENDING[:-1].decode(),
unread_body=UNREAD[:-1].decode())
events = [dict(unit=PROCESS['unit'], pid=55, boot_id=PROCESS['boot_id'], at_ns=150,
message=f'delivery sequence={seq} sha256={hashlib.sha256(payload).hexdigest()} '
f'status=ack committed_through={seq} elapsed_us=1')
for seq, payload in ((4, batch()), (5, batch(5, body=UNREAD,
offset=len(PENDING + UNREAD))))]
return before, installed, drained, queries, events
def test_clean_raw_snapshot_and_two_line_conservation(self):
chain = self.chain()
self.assertEqual(chain[0]['node_id'], NODE.hex())
self.assertEqual(chain[0]['ack'], 3)
self.assertEqual(chain[0]['sources'][0]['sha256'], hashlib.sha256(batch()).hexdigest())
self.assertEqual(chain[0]['cursors'][SOURCE]['offset'], len(PENDING))
self.module.conserve(*chain)
def test_complete_frame_and_envelope_refusals(self):
wrong_wire = ld(1, b'\x01') + batch()[2:]
variants = [batch(version=2), wrong_wire, batch(node=NODE[:-1]),
batch(generation=0), batch(sequence=0), batch(sequence=2**64 - 1),
batch(generation=2**64), batch(cursor_bytes=cursor(len(PENDING), skipping=2))]
for index, payload in enumerate(variants):
with self.subTest(payload=index), self.assertRaises(ValueError):
self.snapshot(f'envelope-{index}', [payload])
changes = [lambda root: (root / 'batches.faj').write_bytes(frame(batch())[:-1]),
lambda root: (root / 'batches.faj').write_bytes(frame(batch()) + b'F'),
lambda root: (root / 'batches.faj').write_bytes(frame(batch())[:16] + b'X' + frame(batch())[17:]),
lambda root: (root / 'recovery-required').write_bytes(b'failure'),
lambda root: (root / 'append-in-progress').write_bytes(b'interrupted'),
lambda root: (root / 'acked').write_bytes(b'FAK1' + struct.pack('<QQ', 1, 2**64 - 1)),
lambda root: (root / 'acked').write_bytes(b'FAK1' + struct.pack('<QQ', 2, 0))]
for index, change in enumerate(changes):
with self.subTest(frame=index), self.assertRaises(ValueError):
self.snapshot(f'frame-{index}', change=change)
def test_nested_otlp_structural_refusals(self):
# Prost permits a sparse record with a valid unknown field, not malformed wire keys.
positive = self.snapshot('otlp-unknown-valid',
[batch(logs_bytes=ld(1, ld(2, ld(2, vi(99, 0)))))])
self.assertEqual(positive['sources'][0]['rows'][0]['body'], '')
variants = (
('field-zero', b'\x00\x00', 'protobuf (field|tag)'),
('field-too-large', ld(1, ld(2, ld(2, varint((1 << 29) << 3) + b'\x00'))),
'protobuf (field|tag)'),
('varint-too-long', ld(1, ld(2, ld(2, varint(99 << 3) + b'\x80' * 10 + b'\x00'))),
'varint'),
)
for name, logs, reason in variants:
with self.subTest(defect=name), self.assertRaisesRegex(ValueError, reason):
self.snapshot('otlp-' + name, [batch(logs_bytes=logs)])
def test_generic_reader_positive_forms(self):
start, end = varint(99 << 3 | 3), varint(99 << 3 | 4)
variants = (vi(1, 7) + ld(99, b'\x00\xff'),
b'\x08\x87\x00', vi(1, 7) + vi((1 << 29) - 1, 2**64 - 1),
vi(1, 7) + start + ld(98, b'\x00\xff') + end,
vi(1, 7) + start * 100 + end * 100)
for index, raw in enumerate(variants):
with self.subTest(form=index):
self.assertEqual(self.module.one(self.module.fields(raw, {1: 0}), 1), 7)
def test_generic_reader_structural_refusals(self):
start, end = varint(99 << 3 | 3), varint(99 << 3 | 4)
variants = (
('zero', b'\x00\x00', 'protobuf (field|tag)'),
('large-key', varint((1 << 29) << 3) + b'\x00', 'protobuf (field|tag)'),
('wire', varint(99 << 3 | 6), 'wire'),
('long-value', varint(99 << 3) + b'\x80' * 10 + b'\x00', 'varint'),
('overflow', varint(99 << 3) + b'\xff' * 9 + b'\x02', 'varint|overflow'),
('long-length', varint(99 << 3 | 2) + b'\x80' * 10 + b'\x00', 'varint'),
('length', varint(99 << 3 | 2) + varint(3) + b'ab', 'length|truncat'),
('fixed', varint(99 << 3 | 1) + b'\x00' * 7, 'fixed|truncat'),
('lone-end', end, 'group'), ('unterminated', start + vi(98, 0), 'group'),
('mismatch', start + varint(98 << 3 | 4), 'group'),
('depth', start * 101 + end * 101, 'depth|recursion'),
('unknown-at-limit', start * 100 + vi(98, 0) + end * 100, 'depth|recursion'),
)
for name, raw, reason in variants:
with self.subTest(defect=name), self.assertRaisesRegex(ValueError, reason):
self.module.fields(raw, {})
def test_packed_numeric_reader_forms_and_refusals(self):
# The reader spec names element wire-kind, not an opaque bytes-as-message guess.
for wire, width in ((1, 8), (5, 4)):
for raw in (varint(1 << 3 | wire) + b'\x00' * width,
ld(1, b'\x00' * (2 * width)), ld(1, b'')):
with self.subTest(wire=wire, bytes=len(raw)):
self.assertEqual(self.module.one(self.module.fields(
raw + vi(2, 7), {1: ('packed', wire), 2: 0}), 2), 7)
with self.subTest(width=width), self.assertRaisesRegex(ValueError, 'packed|width'):
self.module.fields(ld(1, b'\x00' * (width - 1)), {1: ('packed', wire)})
for raw in (ld(1, varint(2**64 - 1) + b'\x80\x00'), vi(1, 2**64 - 1)):
with self.subTest(varint=raw):
self.module.fields(raw, {1: ('packed', 0)})
with self.assertRaisesRegex(ValueError, 'varint'):
self.module.fields(ld(1, b'\x80' * 10 + b'\x00'), {1: ('packed', 0)})
def test_logs_common_schema_and_utf8_refusals(self):
variants = (
('record-wire', export_record(vi(3, 0)), 'wire'),
('body-string', export_record(ld(5, ld(1, b'\xff'))), 'UTF|utf'),
('event-name', export_record(ld(12, b'\xff')), 'UTF|utf'),
('resource-key', export_record(b'', resource=ld(1, ld(1, b'\xff'))), 'UTF|utf'),
('scope-name', export_record(b'', scope=ld(1, b'\xff')), 'UTF|utf'),
('entity-type', export_record(b'', resource=ld(3, ld(2, b'\xff'))), 'UTF|utf'),
('array-body', export_record(ld(5, ld(5, ld(1, ld(1, b'\xff'))))), 'UTF|utf'),
('kv-list', export_record(ld(5, ld(6, ld(1, ld(1, b'\xff'))))), 'UTF|utf'),
('scope-url', ld(1, ld(2, ld(2, b'') + ld(3, b'\xff'))), 'UTF|utf'),
('resource-url', ld(1, ld(2, ld(2, b'')) + ld(3, b'\xff')), 'UTF|utf'),
)
for name, logs, reason in variants:
with self.subTest(defect=name), self.assertRaisesRegex(ValueError, reason):
self.snapshot('schema-' + name, [batch(logs_bytes=logs)])
def test_logs_common_and_message_depth_boundaries(self):
kv = ld(1, b'key') + ld(2, ld(1, b'value')) + vi(3, 2**64 - 1)
resource = ld(1, kv) + vi(2, 0) + ld(3, ld(1, b'url') + ld(2, b'type')
+ ld(3, b'id') + ld(4, b'description'))
scope = ld(1, b'scope') + ld(2, b'version') + ld(3, kv) + vi(4, 0)
for index, value in enumerate((vi(2, 2), ld(7, b'\x00\xff'), vi(8, 2**64 - 1),
ld(5, ld(1, ld(1, b'array'))), ld(6, ld(1, kv)))):
with self.subTest(value=index):
got = self.snapshot(f'common-{index}', [batch(logs_bytes=export_record(
ld(5, value), resource=resource, scope=scope))])
self.assertEqual(got['sources'][0]['rows'][0]['body'], '')
# Export→ResourceLogs→ScopeLogs→LogRecord→AnyValue starts depth4.
for layers, leaf, accepted in ((48, ld(1, b'deep'), True),
(49, ld(1, b'deep'), False), (48, vi(99, 0), False)):
value = leaf
for _ in range(layers):
value = ld(5, ld(1, value)) # AnyValue.array→ArrayValue.values→AnyValue.
payload = batch(logs_bytes=export_record(ld(5, value)))
with self.subTest(layers=layers, unknown=leaf[:1]):
if accepted:
self.assertEqual(self.snapshot('depth-message-positive', [payload])
['sources'][0]['rows'][0]['body'], '')
else:
with self.assertRaisesRegex(ValueError, 'depth|recursion'):
self.snapshot(f'depth-message-{layers}-{leaf[0]}', [payload])
def test_metrics_traces_container_scope(self):
# Literal host fields: CPU cumulative Sum and memory Gauge; no Rust codec used.
times = b'\x11' + struct.pack('<Q', 120) + b'\x19' + struct.pack('<Q', 123)
gauge = ld(1, b'system.memory.total') + ld(3, b'By') + ld(5, ld(1,
times + b'\x31' + struct.pack('<q', 4096)))
total = ld(1, b'system.cpu.time') + ld(3, b's') + ld(7, ld(1,
times + b'\x21' + struct.pack('<d', 1.25)) + vi(2, 2) + vi(3, 1))
metrics = ld(1, ld(2, ld(2, gauge) + ld(2, total)))
span = ld(1, b'\x01' * 16) + ld(2, b'\x02' * 8) + ld(5, b'literal-span')
payload = batch() + ld(5, metrics) + ld(9, export_record(span))
got = self.snapshot('host-gauge-sum-trace', [payload])
self.assertEqual(got['sources'][0]['payload'], payload)
self.assertEqual(got['sources'][0]['rows'][0]['body'], PENDING[:-1].decode())
variants = (
('metric-export-length', 5, b'\x0a\x03ab', 'length|truncat'),
('trace-export-varint', 9, varint(99 << 3) + b'\x80' * 10 + b'\x00', 'varint'),
('metric-resource-field', 5, ld(1, b'\x00\x00'), 'protobuf (field|tag)'),
('trace-scope-tag', 9, ld(1, ld(2, varint((1 << 29) << 3) + b'\x00')), 'protobuf (field|tag)'),
('metric-record-field', 5, export_record(b'\x00\x00'), 'protobuf (field|tag)'),
('trace-record-field', 9, export_record(b'\x00\x00'), 'protobuf (field|tag)'),
('metric-container-wire', 5, vi(1, 0), 'wire'),
)
for name, signal, raw, reason in variants:
with self.subTest(defect=name), self.assertRaisesRegex(ValueError, reason):
self.snapshot(name, [batch() + ld(signal, raw)])
# These malformed nested leaves must remain opaque to this limited witness.
# Product ingest must independently reject them; this is not semantic admission.
opaque = batch() + ld(5, export_record(ld(5, b'\xff'))) + ld(9, export_record(ld(11, b'\xff')))
self.assertEqual(self.snapshot('opaque-product-leaves', [opaque])['sources'][0]['payload'], opaque)
def test_original_payload_log_projection_binding(self):
stamp = b'\x59' + struct.pack('<Q', 123)
unknown = varint(99 << 3 | 3) + ld(98, b'\x00\xff') + varint(99 << 3 | 4)
variants = (
('unknown-group', stamp + ld(5, ld(1, b'kept')) + unknown, 'kept', {}),
('last-bool', stamp + ld(5, ld(1, b'discarded') + vi(2, 1)), '', {}),
('last-string', stamp + ld(5, vi(2, 1) + ld(1, b'kept')), 'kept', {}),
('merged-body', stamp + ld(5, ld(1, b'kept')) + ld(5, vi(99, 0)), 'kept', {}),
('merged-value', stamp + ld(5, ld(1, b'kept')) + ld(6, ld(1, b'key')
+ ld(2, ld(1, b'value')) + ld(2, b'')), 'kept', {'key': 'value'}),
)
for name, record, body, attrs in variants:
original = batch(logs_bytes=export_record(record)) + ld(99, b'\x00\xff')
with self.subTest(projection=name):
source = self.snapshot('projection-' + name, [original])['sources'][0]
self.assertEqual((source['payload'], source['sha256']),
(original, hashlib.sha256(original).hexdigest()))
self.assertEqual(source['rows'], [dict(node='spindle-1', node_id=NODE.hex(),
sequence=4, index=0, observed_ns=123, body=body, attributes=attrs)])
def test_full_cursor_prefix_identity_and_sequence_order(self):
older = (vi(1, 1) + ld(2, NODE) + vi(3, 1) + vi(4, 4)
+ ld(7, cursor(len(PENDING + UNREAD))) + ld(8, b'older cursor fixture'))
snapshot = self.snapshot('latest', [older, batch(5)])
self.assertEqual(snapshot['cursors'][SOURCE]['offset'], len(PENDING))
self.assertEqual(snapshot['cursors'][SOURCE]['prefix_crc'], zlib.crc32(PENDING))
for index, raw in enumerate((cursor(len(PENDING), prefix_crc=1),
cursor(len(PENDING), skipping=1),
cursor(len(PENDING), identity=ld(1, b'x') + vi(2, 1)))):
with self.subTest(cursor=index), self.assertRaises(ValueError):
self.snapshot(f'cursor-{index}', [batch(cursor_bytes=raw)])
self.logs[SOURCE]['btrfs_identity'] = dict(uuid=NODE.hex(), subvolume_id=7)
full = ld(1, NODE) + vi(2, 7)
valid = self.snapshot('btrfs', [batch(cursor_bytes=cursor(len(PENDING), identity=full))])
self.assertEqual(valid['cursors'][SOURCE]['btrfs_identity'], self.logs[SOURCE]['btrfs_identity'])
def test_stop_and_path_admission_refusals(self):
mutations = [lambda receipt: receipt.update(stop_confirmed=False),
lambda receipt: receipt.update(extra_project_units=['unexpected.service']),
lambda receipt: receipt['units'].pop('fabrico11y-node.service'),
lambda receipt: receipt['units']['fabrico11y-node.service'].update(main_pid=55),
lambda receipt: receipt['units']['fabrico11y-node.service'].update(query_exit=1)]
for index, mutation in enumerate(mutations):
with self.subTest(stop=index), self.assertRaises(ValueError):
self.snapshot(f'stop-{index}', receipt_change=mutation)
target = self.root / 'outside-identity'; target.write_bytes(b'FAI1' + NODE + struct.pack('<Q', 1))
def symlink(root):
(root / 'identity').unlink(); (root / 'identity').symlink_to(target)
with self.assertRaises(ValueError): self.snapshot('symlink', change=symlink)
self.assertEqual(target.read_bytes(), b'FAI1' + NODE + struct.pack('<Q', 1))
def test_install_ack_query_and_reboot_conservation_refusals(self):
chain = self.chain()
mutations = [lambda args: args[0].update(ack=4),
lambda args: args[1]['binding'].update(run_id='other-run'),
lambda args: args[1]['binding'].update(candidate_package_sha256='00' * 32),
lambda args: args[1].update(node_id='00' * 16),
lambda args: args[1]['sources'].clear(),
lambda args: args[1]['sources'][0].update(sha256='00' * 32),
lambda args: args[1]['cursors'][SOURCE].update(offset=len(PENDING + UNREAD)),
lambda args: args[2].update(ack=0),
lambda args: args[4].clear(),
lambda args: args[4][0].update(pid=44),
lambda args: args[4][0].update(unit='other.service'),
lambda args: args[4][0].update(boot_id='predecessor-boot'),
lambda args: args[4][0].update(at_ns=99),
lambda args: args[4][0].update(message=args[4][0]['message'].replace(hashlib.sha256(batch()).hexdigest(), '00' * 32)),
lambda args: args[3]['after_drain'].append(copy.deepcopy(args[3]['after_drain'][0])),
lambda args: args[3]['after_reboot'][0].update(observed_ns=124),
lambda args: args[3].update(boot_after='literal-boot-a')]
for index, mutation in enumerate(mutations):
args = copy.deepcopy(chain); mutation(args)
with self.subTest(conservation=index), self.assertRaises(ValueError):
self.module.conserve(*args)
if __name__ == '__main__':
unittest.main()
docs/experiments/formal/installation-release-protocol.md
# Registered release package and lifecycle checks
Status: revision 4 registered 2026-10-12 for the frozen candidate: its exact
candidate and L1 predecessor versions and the dated Debian image. Revision 3,
registered 2026-10-10 before stopped-overlay continuation, named the earlier
candidate; a result belongs to the revision it ran under. This extends the
[baseline installation protocol](installation-acceptance-protocol.md) for the
owner-selected Debian/Fedora release matrix. No result is recorded here.
These checks are finite local guest measurements, not deployment qualification.
## Stopped-overlay continuation
A separately recorded continuation may finish L2/removal from an immutable
stopped trial whose historical baseline exited 0 and whose exact candidate
L1 upgrade passed, but whose reboot observer failed. Preserve the original
overall failure. Bind its receipt/acceptance/harness hashes, exact candidate and
predecessor, source identity, stopped overlay SHA-256 and official backing image.
Use a new owned overlay with that stopped disk as read-only backing; never
mutate the retained input. Freeze the saved pre-reboot boot ID, config hashes,
account identity and exact row witnesses from that disk before L2. A genuinely
different live boot identity remains required; mere SSH reconnection is insufficient.
Use stdin for the privileged boot comparison so the unprivileged SSH shell
cannot expand protected paths. Record each actual phase exit, caps and cleanup
in a new continuation receipt. This does not qualify unrun baseline criteria on
the candidate or relabel the original failed run. The final new candidate still
requires fresh full installation/lifecycle trials.
## Fixed inputs and containment
Freeze source (including Rust crates, UI inputs, vendored dependencies, package
scripts, helpers, license/notice), Rust 1.99.0, Cargo.lock, a successful exact
console build, package SHA-256, and the candidate/predecessor versions before
execution. Candidate package version is `0.1.0~alpha.1` on Debian and
`0.1.0-0.alpha.1` on RPM. L1 predecessors are the registered historical Debian
package re-versioned to `0.1.0~alpha.0` (payload and maintainer scripts unchanged,
`packaging/reversion-predecessor-deb.sh`) and the constructed RPM fixture
`0.1.0-0.alpha.0`. This metadata does not authorize a release tag.
`build-isolated.py --console-build BUILD --family debian|fedora` builds in the
bounded rootless Debian 12 sysroot; RPM construction uses `rpmbuild` there too,
keeping the same registered glibc floor. Copy verified UI assets into owned
storage; recheck their receipt/hashes in the frozen-source guest before packaging.
A source or asset mismatch blocks construction.
The candidate RPM may instead use `build-rpm.sh OUT CANDIDATE_DEB
--candidate-deb BUILD_RECEIPT [--mutate=NAME]`. Require a completed successful
Debian build/cleanup receipt, its exact package SHA-256, exact candidate Debian metadata,
and its matching frozen-source manifest/bundle hashes. Preserve every regular
Debian data member unchanged, including native binaries, UI assets and notices;
record their hashes plus source and input package identity in RPM documentation.
Use the frozen candidate account-check script for RPM installation. RPM metadata,
scriptlets, dependencies and SELinux context restoration remain Fedora-specific.
Reject historical/wrong-version/substituted payloads before staging. This is
repackaging the same Debian12-sysroot binaries, not an independent Fedora compiler
result. Actual Fedora enforcing-mode lifecycle cells still remain necessary.
Run every host command through `python3 -B tools/resource_group.py -- COMMAND`.
Use only the mounted data drive, a maximum 20 GiB/no-swap descendant cgroup and
the default 30-minute deadline. The package-build container inherits that cgroup;
QEMU is its descendant. One guest at a time, 2 virtual CPUs, 6144 MiB guest RAM,
8 GiB virtual disk, disk-backed owned overlay/scratch, no KVM requirement. Preserve
failures/receipts and stop/reap guest processes before cleanup. Enforce the
existing 100 GB data-tree budget, stop admission at 95 GB including the reserved
8 GiB build or full guest disk plus image/log overhead, and record free disk
before admission. Do not
interpret these guest limits or a tiny fixture as the 100 GB retention-capacity
release result.
Debian: Debian 13 genericcloud amd64 build 20261001-2618/systemd 257, from that
build's [dated directory](https://cloud.debian.org/images/cloud/trixie/20261001-2618/),
SHA-512 `f46f0671…b468c3d` as pinned in the harness. Fedora: Cloud Base Generic 44-1.7 x86_64, from the
[official Fedora image directory](https://dl.fedoraproject.org/pub/fedora/linux/releases/44/Cloud/x86_64/images/),
583729152 bytes, SHA-256
`28680fe5b371a5a82ebf43a31926e086a168e59949d03969c5093e7071f90b7f`.
Match the official HTTPS checksum manifest and the downloaded image before boot.
The Fedora checksum manifest is signed; this harness pins its reviewed HTTPS
checksum and does not claim independent OpenPGP signature validation. Record
actual guest OS, systemd, kernel, PID1, cgroup controllers and SELinux mode.
Fedora must remain enforcing; no permissive switch or broad allow policy is a
workaround. An unavailable tool, image mismatch or unavailable gate is not a pass.
## CI host containment
GitHub-hosted runner bootstrap binds disk-backed `/mnt` to the required data
mount and starts the user systemd manager. Before project children run, configure
`user-UID.slice` as a stricter ancestor: memory maximum is the lesser of 12 GiB
and 75% of actual MemTotal, high is five-sixths of that maximum, swap maximum
zero, CPU quota no greater than the available CPU count. Verify the actual
kernel memory/high/swap/CPU files and write `ci-parent.json`; a missing or
mismatching bound fails closed. `RUST_TEST_THREADS=2` and `CARGO_BUILD_JOBS=2`
pass through the launcher. This hosted-runner ancestor does not alter the local
20 GiB contract or establish release workload acceptance.
## Gates
The existing A1–A12 criteria remain unchanged. Backend package operations are
`dpkg` on Debian and `rpm` on Fedora; guest extraction uses `rpm2cpio/cpio`.
The fixture's explicit 20 GiB retention policy is a narrow delivery/lifecycle
fixture. The release's 100 GB default is independently checked in configuration
and examples, and needs the release storage workload before capacity acceptance.
| Gate | Pass rule |
| --- | --- |
| A2c/A2d | Fresh candidate install refuses duplicate UID0 and GID0 named fabricolly before unpack. Preserve root account/group; delete only disposable duplicate fixture entries. Historical predecessor migration cells do not claim this corrected admission behavior. |
| P1 | Both exact packages contain a complete hash-verified console (HTML, compiled WASM, manifest, normal/maskable icons, worker), CSP metadata, binaries, units, examples and Apache-2.0 license/notice. Reject changed/missing/unlisted/symlink assets, a failed build receipt and a mismatching source identity. |
| F1/F2 | Fedora reports SELinux Enforcing before delivery and after restart/removal; every A1–A12 delivery/sandbox/custody criterion still succeeds. |
| F3 | Boot audit contains no AVC/USER_AVC with a Fabric binary `comm`; retain audit and service journal. An unavailable audit facility is a failed prerequisite. |
| L1 | Install an exact strictly older predecessor, complete A1–A12, upgrade to the candidate with the native package manager, restart services, and preserve exact three fixture rows including source/sequence/index/timestamps/body/attributes, config checksums and service-account identity. Validate installed UI hashes after upgrade. An equal-version reinstall is refused. |
| L2 | Record the guest boot ID, reboot the VM, reconnect only after a distinct boot ID, and observe both enabled services active. Exact retained rows/config/account still match. Fedora remains enforcing. |
| A13a | Native removal stops both services and removes binaries while retaining telemetry/configuration/account. |
| A13b | Debian native purge removes state/configuration. RPM has no purge operation: verify removal preserved both, then explicitly delete the disposable fixture state/configuration. This cleanup is not presented as a package purge guarantee. |
| F4 | Fedora remains enforcing through reboot/removal; post-reboot audit contains no Fabric-process AVC. |
L1's expected bodies are `accept-line-before-start`, `accept-line-during-outage`
and `accept-line-after-pressure`, each exactly once from `spindle-1`. Identity
comparison includes every projected field and uses Python JSON integers. The
query fixture intentionally has a denied-log gap; these lifecycle checks do not
claim full-source completeness or change the independent query oracle.
These are **explicit legacy-admin migration fixtures**. Fresh candidate tests use
`--legacy-server`; upgrade installs a guest-only systemd drop-in selecting
`serve-legacy`. The shipped normal service uses passkey configuration, and the
separate authenticated-console protocol must exercise its production entrypoint.
Legacy-admin success cannot satisfy access-control or authenticated UI gates.
Use `run-qemu.py --family FAMILY --package CANDIDATE --source-commit COMMIT
--run-id ID --memory-stressor parallel --upgrade-from PREDECESSOR` for L1/L2.
Without `--upgrade-from`, the harness runs the ordinary baseline/removal path and
has no upgrade/reboot result. The previous package's hash/version belongs in the
receipt. A synthetic predecessor must be labeled as such and cannot establish
migration from a historical implementation it did not contain.
## Negative controls and evidence
### Prospective removal-checker counterexample controls
Coordination C-035 / C-aud-7 minimized an unenforced-check defect at harness
`de77de7399710c5afd50ae0de981fda6e0b75f30`: negated commands and the first
member of `&&` lists bypass Bash errexit; baseline A13a accepted one live unit.
`InstallationShellControls` in `tools/qualification/install/test_run_qemu.py`
invokes the actual lifecycle remove phase with isolated paths and package/service
stubs. Either active, activating, deactivating or reloading unit, literal Fabric
AVC, missing retained state and incomplete purge must exit1 without the affected
PASS; clean fixtures must exit0. Baseline A13a rejects the same live states.
Service-query controls additionally require both observed states: empty or one-line
output and exit1 despite two inactive lines fail A13a. A literal grep exit2 at the
post-reboot AVC check fails F4; clean inactive queries and readable empty audits
remain positive controls. These are verifier prerequisites, not guest outcomes.
The actual QEMU cleanup AST additionally requires an independently detected RPM mutation
and confirmed process stop; unrelated/provenance failures retain scratch.
These prospective controls precede repair and claim no guest outcome. Preserve
the contained red first; after repair, register its actual fix commit in
`docs/formal/counterexamples.json` in a separate trust-boundary commit. RPM
construction plans explicitly create each fresh mutant output directory before
`build-rpm.sh`; an existing directory is refused. Limits and gate rules remain.
The prerequisite controls ran at `8b5d3329cb03d6c470f24b5dba46fba03121804e`:
five methods exposed exactly seven intended failures, exit1 with confirmed stop
(`installation-prerequisite-checker-red-03.json`, log SHA-256
`accc6c250b667e9a8b86664e4e8d6c354118a730430c7b36fed85ababb93c397`).
Repair `ce387f75610bc4bac82955fbb722c9d61e4fcbc1` keeps those expectations.
On that clean revision, `python3 -B tools/qualification/install/test_run_qemu.py`
ran all nine tests, exit0 with confirmed stop
(`installation-prerequisite-checker-green-03.json`, log SHA-256
`a77f13b9a487d5a928e54cb437e007c20e7a2ec8255b3fdb3bfd1ff37598d6d9`,
stopped resource receipt SHA-256
`5a29865387ea8ff2778e38900b5c27f8db3411b3873306227e1ddbe3d629fc40`).
`bash -n tools/qualification/install/acceptance.sh tools/qualification/install/lifecycle.sh`
and pinned Bun1.4.0 `tools/docs/check.mjs` also exited0 with confirmed stop on
that revision. The earlier bare-Bun attempt exited2 because Bun was unavailable;
its failure evidence remains preserved. These finite checker results establish
no new guest or package outcome.
### Prospective stopped-custody upgrade witness
Issue #41's Spool/backlog upgrade obligation needs a predecessor-produced
unacknowledged Batch and a distinct unread newline-terminated source suffix
across stopped native install, then exact candidate ACK/query and real reboot.
Historical L1/L2 checks remain required. A retained file hash alone cannot
establish recoverable delivery; shared-Rust `spool_dump` is a diagnostic.
`test_custody_witness.py` independently hand-encodes small FAB1/FAC1/protobuf
fixtures for prospective `custody_witness.stopped_snapshot` and `conserve`.
The initial named `MechanismAvailability` failure records absence of the
adapter; it is not a product defect or corrupt-frame counterexample. Actual
specific refusals remain NOT RUN until that adapter exists. Its unchanged
`CustodyWitnessControls` require complete CRC-checked frames, envelope/wire/u64
bounds, nonterminal sequence/ACK, full Cursor/Btrfs identity and consumed-prefix
CRC, last cursor in sequence order rather than maximum offset, complete stopped
service observations, no-follow owned paths, and exact two-line custody joins.
Reset source identity, missing/changed pending bytes, premature ACK, duplicated
or changed query rows and unchanged reboot identity must be refused. ACK events
must match the independently observed candidate node PID/unit/boot/run window
and exact raw Batch hash; their message alone omits Strand identity.
Run only through the admitted serial resource launcher: default1800 outer and
1GiB reservation for synthetic controls. Native cells keep1800 outer,180-second
lifecycle phases and9GiB whole guest reservation. Preserve failures and raw
source custody, exporting only hashes of credentials. Frozen packages, wire
semantics, query/delivery/rate oracles and all existing gate limits stay fixed.
This prospective registration claims no checker or guest acceptance.
#### Nested OTLP structural amendment, 2026-10-13
Preregister `CustodyWitnessControls.test_nested_otlp_structural_refusals` with
hand-encoded inner OTLP logs inside CRC-valid FAB1/FAC1 frames. A sparse LogRecord
containing valid unknown field99 is a positive, even though its projected body is
empty. ExportLogsServiceRequest field0 (`00 00`), a nested LogRecord field number
greater than `(1<<29)-1`, and an eleven-byte nested varint must raise `ValueError`
for the malformed protobuf field/tag or varint; accepting an empty projection or
rejecting only missing sentinels is insufficient. The locked prost0.14.4
`encoding::decode_key` bounds the key to u32 and tags to1 through `(1<<29)-1`, and
its varint decoder rejects values requiring more than ten bytes. The existing
Rust `Batch::validate` decodes each OTLP signal before admitting its bytes.
These source constraints add no new wire semantics or oracle changes. Run the
named control only through the registered default1800s/full1GiB serial driver;
retain the actual red before a separately reviewed structural-validator repair.
#### Bounded reader and original-byte projection amendment, 2026-10-13
Coordination C-061/X-106 admits a 420-line production ceiling, at most +120 after
the 297-line adapter, with no oracle or limit change. Register seven additional
`CustodyWitnessControls` methods before repair; the three malformed cases and
sparse positive in `test_nested_otlp_structural_refusals` stay unchanged.
`test_generic_reader_positive_forms` admits opaque unknown bytes, nonminimal
valid varints, the largest legal tag/u64 and matched unknown groups, including
100 empty nested groups. `test_generic_reader_structural_refusals` rejects
invalid keys/wires, truncation, overlong/overflowing value and length varints,
lone/mismatched/unterminated groups, 101 nested groups and an unknown field at
exhausted depth. Controls call the existing `fields(raw, types)` reader directly.
`test_packed_numeric_reader_forms_and_refusals` registers its explicit
`('packed', element_wire_kind)` spec for kinds 0/1/5: both unpacked and packed
forms are admitted; malformed packed varints and nondivisible fixed-8/fixed-4
payloads are refused. Arbitrary opaque bytes are never guessed to be messages.
`test_logs_common_schema_and_utf8_refusals` requires declared wires and UTF-8
through ExportLogsServiceRequest/ResourceLogs/ScopeLogs/LogRecord and all
consumed common schemas: Resource, InstrumentationScope, EntityRef, KeyValue,
AnyValue, ArrayValue and KeyValueList. `test_logs_common_and_message_depth_boundaries`
admits literal common fields, opaque bytes and nonzero bools. From the export
root, 48 AnyValue.array/ArrayValue.values pairs place a known string at depth 100
and remain valid; 49 pairs or an unknown scalar at exhausted depth fail.
Each signal export has its own recursion budget, matching pinned prost 0.14.4.
`test_metrics_traces_container_scope` requires strict Export→Resource→Scope→
Metric/Span wire containers; malformed keys/varints/lengths and container wires
fail. Hand-encoded `system.memory.total` Gauge and `system.cpu.time` cumulative
Sum use the actual host shapes in `src/spindle/runtime.rs:266-299` and
`crates/fabric-adapter-linux/src/host.rs:190-217`; a literal Span is also positive.
Metric/Span nested leaves remain opaque, explicitly witnessed by a malformed
nested Gauge/Event fixture that this reader admits without asserting semantic
validity. Product ingest decoding is the leaf-semantics authority; frame CRC
establishes byte integrity, not semantic validity. No general prost parity or
native product acceptance is claimed by these finite controls.
`test_original_payload_log_projection_binding` registers exact literal log rows
for valid unknown groups, last oneof arm and merged optional body/value messages.
The stored original Batch payload and SHA-256 must remain exact. A separately
filtered log projection may feed the byte-identical query oracle; it cannot
replace raw custody or exact-byte candidate ACK evidence. The caller projects
log rows only and must not reinterpret opaque metric/trace leaves through the
oracle merely to obtain those rows. Query/delivery/rate oracles remain unchanged;
native caller authentication against immutable candidate receipts stays separate.
Run these source-only fixtures through the same serial default 1800s/full 1GiB
driver; preserve actual red before production repair and rerun unchanged controls.
Refusals name the violated wire/string/group/depth property. Legitimate-form
rejections or wrong projections on the old reader are harness defects, not
product qualification failures. No execution result is recorded by this amendment.
Keep the existing named Debian package defects: root-user must fail A6,
no-collision-check must fail A2a/A2b, and no-memory-max must fail A8. The existing
mutation grader must reject unrelated failures and bad provenance/environment.
Fedora package mutations are separate exact RPM artifacts built with
`build-rpm.sh OUT BUILD --mutate=NAME`; grade them with
`run-qemu.py --family fedora --package MUTANT --expected-mutation NAME` and the
same named A6/A2a+A2b/A8 failure requirements. The harness refuses the Debian
repacker on Fedora. An unrelated failure, missing fixture/provenance or NOT-RUN
cannot count as detection. Required RPM negative controls remain unrun until
these artifacts are constructed and independently graded; a baseline success
cannot erase that missing gate.
A Fedora migration predecessor can be constructed by
`build-rpm.sh OUT HISTORICAL_DEB --historical-deb`. Its source is the exact
`readiness-debian12-03` alpha1 Debian artifact, SHA-256
`ac71995b3c8b88cbb6f7dd05d9854e347b7a783caadc1b602f6590d6460f4a2f`.
Preserve its historical binaries, static units, examples, sysusers and account
collision preinstall script. The fixture's `MIGRATION-FIXTURE.json` records
those payload hashes; native RPM lifecycle scripts adapt package operations.
It contains no historical console and is a **constructed predecessor fixture**,
not evidence that an alpha1 Fedora package shipped or was accepted. L1 upgrades
its historical implementation/state into the exact current candidate RPM.
Every receipt records the exact package/helper/image/source identities, command,
exit, chosen A12 fixture, measured resources, real lifecycle stages and cleanup.
A failed, interrupted or missing required phase blocks the corresponding cell.
First-run results and limitations belong in a separate result record.
Exact driver
import datetime, hashlib, json, os, re, signal, subprocess, sys, time
from pathlib import Path
ROOT = Path('/run/media/kmosoti/data/FabricO11y')
REPO = ROOT / 'scratch/installation-custody-witness'
OUT = ROOT / 'results/service-opening-candidate'
STORE = ROOT / 'scratch/service-opening-census'
FROZEN = '353455b8a90429357d9bc198152e19d0bc8fb3f0'
def require(value, reason):
if not value:
raise ValueError(reason)
label, reserve_text, *command = sys.argv[1:]
reserve = int(reserve_text)
delegated = command[:1] == ['--delegate']
if delegated: command = command[1:]
require(re.fullmatch(r'[a-z0-9-]{1,80}', label) and reserve > 0 and command, 'ordinary workload admission guard')
record_path = OUT / (label + '.json')
require(not record_path.exists(), 'unique run label required')
def census():
# Rootless container files can belong to subordinate UIDs. Inspect all of
# them in the same user namespace; never skip unreadable directories.
require(not STORE.is_symlink(), 'ordinary workload admission guard')
require((STORE / '.owned-census-store').read_text() == 'service-opening-census-v1\n', 'ordinary workload admission guard')
args = ['podman', '--root', str(STORE/'graph'), '--runroot', str(STORE/'run'),
'--tmpdir', str(STORE/'tmp'), '--storage-driver', 'vfs',
'unshare', 'python3', '-B', str(OUT/'census.py')]
env = dict(os.environ, TMPDIR=str(STORE/'tmp'), TMP=str(STORE/'tmp'), TEMP=str(STORE/'tmp'))
result = subprocess.run(args, env=env, text=True, capture_output=True, timeout=25)
if result.returncode:
raise RuntimeError(f'strict namespace census failed ({result.returncode}): {result.stderr}')
observed = json.loads(result.stdout)
require(observed['uid'] == 0 and observed['gid'] == 0, 'ordinary workload admission guard')
require(type(observed['data_bytes']) is int and observed['data_bytes'] >= 0, 'ordinary workload admission guard')
return observed
def git(*args):
return subprocess.check_output(['git', *args], cwd=REPO, text=True).strip()
require(git('rev-parse', 'HEAD') == FROZEN and not git('status', '--porcelain'), 'clean frozen source required')
require(not subprocess.check_output(['systemctl', '--user', 'list-units', 'fabric-work-*', '--state=running', '--plain', '--no-legend'], text=True).strip(), 'another project workload is active')
initial = census()
used = initial['data_bytes']
record = {'label': label, 'command': command, 'delegated': delegated, 'frozen': None, 'scope': 'harness controls only; no candidate acceptance evidence', 'harness_revision': FROZEN, 'source_status_before': git('status', '--porcelain'), 'storage_root': str(ROOT), 'admission_bytes': used, 'admission_observation': initial, 'reserved_bytes': reserve, 'observations': [], 'started_utc': datetime.datetime.now(datetime.timezone.utc).isoformat()}
if used + reserve > 95_000_000_000:
record.update(state='not_run', reason='95 GB storage admission threshold')
record_path.write_text(json.dumps(record, indent=2)+'\n')
raise SystemExit('storage admission denied; project command not launched')
receipts = REPO / 'target/resource-containment/runs'
before = set(receipts.glob('*.json'))
env = dict(os.environ, FABRIC_STORAGE_ROOT=str(ROOT), CARGO_HOME=str(ROOT/'toolchain-cache/cargo'), RUSTUP_HOME=str(ROOT/'toolchain-cache/rustup'), CARGO_BUILD_JOBS='2')
env['PATH'] = str(ROOT/'tools/candidate-dpkg/bin') + ':' + env['PATH']
log_path = OUT / (label + '.log')
record.update(state='running', log=str(log_path))
record_path.write_text(json.dumps(record, indent=2)+'\n')
def find_unit():
with log_path.open() as stream:
prefix = stream.read(65536)
match = re.search(r'Running as unit: (fabric-work-[0-9a-f]+)\.service', prefix)
return match.group(1) if match else None
def stop_owned(process):
unit = find_unit()
if unit:
stopped = subprocess.run(['systemctl', '--user', 'stop', unit],
text=True, capture_output=True, timeout=25)
record['stop'] = {'unit': unit, 'exit': stopped.returncode, 'stderr': stopped.stderr}
if stopped.returncode:
# Interrupt the launcher too; its finally block independently
# confirms that this exact service has no surviving descendants.
process.send_signal(signal.SIGINT)
else:
process.send_signal(signal.SIGINT)
record['stop'] = {'unit': None, 'launcher_interrupted': True}
with log_path.open('wb') as log:
launcher = ['python3', '-B', 'tools/resource_group.py', *(['--delegate'] if delegated else []), '--', *command]
process = subprocess.Popen(launcher, cwd=REPO, env=env, stdout=log, stderr=subprocess.STDOUT)
next_census = time.monotonic() + 30
try:
while process.poll() is None:
time.sleep(1)
if time.monotonic() >= next_census:
observed = census()
record['observations'].append({'unix':time.time(), **observed})
record_path.write_text(json.dumps(record, indent=2)+'\n')
if observed['data_bytes'] >= 95_000_000_000:
record['interruption'] = '95 GB threshold reached'
break
next_census = time.monotonic() + 30
except BaseException as error:
record['interruption'] = f'{type(error).__name__}: {error}'
finally:
if record.get('interruption') and process.poll() is None:
try:
stop_owned(process)
except BaseException as error:
record['stop_error'] = f'{type(error).__name__}: {error}'
process.send_signal(signal.SIGINT)
status = process.wait()
record['launcher_exit'] = status
if record.get('interruption'): status = 2
with log_path.open('rb') as stream:
log_digest = hashlib.file_digest(stream, 'sha256').hexdigest()
record.update(state='interrupted' if record.get('interruption') else 'completed', exit=status, source_status_after=git('status', '--porcelain'), finished_utc=datetime.datetime.now(datetime.timezone.utc).isoformat(), log_sha256=log_digest, resource_receipts=[])
archive = OUT/'resource-containment'; archive.mkdir(exist_ok=True)
for p in sorted(set(receipts.glob('*.json')) - before):
value=json.loads(p.read_text())
if value['command'] != command: continue
q=archive/p.name; q.write_bytes(p.read_bytes())
record['resource_receipts'].append({'path':str(q),'sha256':hashlib.sha256(q.read_bytes()).hexdigest(),'exit':value['exit'],'stop_confirmed':value['stop_confirmed'],'retained_failure_evidence':value['retained_failure_evidence']})
if not record['resource_receipts'] or any(not r['stop_confirmed'] or r['exit'] != status for r in record['resource_receipts']):
record['receipt_error'] = 'missing or inconsistent outer resource receipt'
status = 2
record['exit'] = status
record_path.write_text(json.dumps(record, indent=2)+'\n')
print(json.dumps(record, indent=2))
if status: print(log_path.read_text()[-8000:])
raise SystemExit(status)