custody reader source plan 03

Prospective: NOT RUN. Clean head 158a5ec1129e6a9789b3d4271cea0020fecd6ff6.

C-083 admitted the literal prefix red at trust609368a; exactly one ValueError-not-raised failure was preserved at wiki93df129. Separate production158a changes only the reader +30/-22:417total/+120over297. Prefix rows tile every archived final-cursor path exactly, checking body/newline/device/inode; validated record maps are reused for projection. All controls/protocol/oracles remain unchanged. Full install discovery includes the17 custody methods and all existing checker/pressure/continuation methods (expected30total). Candidate bindings/native guest producer remain pending; no raw server storage authentication is claimed.

Root runs four exact serial source cells. Every cell uses the driver, resource_group, a fresh full DATA namespace census, full1GiB reserve,95GB admission including reserve/100GB ceiling,16GiB memory high/20GiBmax/no swap/1800seconds. Dirty/moved source or any project workload refuses admission. Any unexpected failure stops this sequence; retain failures and report exact command/exit/stop. Inspect cleanup and read #59 after each cell. Existing Bun1.4.0/106 individual ignored locked dependency links/cache are reused without installation. No VM, installation, native browser, simulator, Rust/C compile, smoke or FULL is launched.

Pins

• /run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-custody-reader-source-03.py SHA-256 edb3ae6c01fede61324907c917825dd1721b0e37b46bbdadf396cb6c968a3e65.
• /run/media/kmosoti/data/FabricO11y/scratch/installation-custody-witness/tools/qualification/install/custody_witness.py SHA-256 2c500203cc26a474366d4b9daf13d63332fb6454fc94649a00f8dd43eb796ebf.
• /run/media/kmosoti/data/FabricO11y/scratch/installation-custody-witness/tools/qualification/install/test_custody_witness.py SHA-256 fa4782a9ff25e72d91546207681db8e9f96c4b771472bbbcacb60b41f85c4a61.
• /run/media/kmosoti/data/FabricO11y/scratch/installation-custody-witness/docs/experiments/formal/installation-release-protocol.md SHA-256 cec645a8f5f029b828526b852385fd411452480a067fbcf0c746499d0d49cd3d.

Exact commands

python3 -B /run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-custody-reader-source-03.py installation-custody-source-controls-03 1073741824 python3 -B -m unittest discover -s tools/qualification/install -p 'test_*.py' -v
python3 -B /run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-custody-reader-source-03.py installation-custody-docs-check-03 1073741824 env PATH=/run/media/kmosoti/data/FabricO11y/tools/bun-1.4.0:/usr/local/bin:/usr/bin:/bin BUN_INSTALL_CACHE_DIR=/run/media/kmosoti/data/FabricO11y/tools/docs-bun-cache /run/media/kmosoti/data/FabricO11y/tools/bun-1.4.0/bun tools/docs/check.mjs
python3 -B /run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-custody-reader-source-03.py installation-custody-docs-tests-03 1073741824 env PATH=/run/media/kmosoti/data/FabricO11y/tools/bun-1.4.0:/usr/local/bin:/usr/bin:/bin BUN_INSTALL_CACHE_DIR=/run/media/kmosoti/data/FabricO11y/tools/docs-bun-cache /run/media/kmosoti/data/FabricO11y/tools/bun-1.4.0/bun tools/docs/check.test.mjs
python3 -B /run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-custody-reader-source-03.py installation-custody-docs-hooks-03 1073741824 env PATH=/run/media/kmosoti/data/FabricO11y/tools/bun-1.4.0:/usr/local/bin:/usr/bin:/bin BUN_INSTALL_CACHE_DIR=/run/media/kmosoti/data/FabricO11y/tools/docs-bun-cache python3 -B tools/docs/test_hooks.py

Full production delta

diff --git a/tools/qualification/install/custody_witness.py b/tools/qualification/install/custody_witness.py
index 6db3fce..16831f0 100644
--- a/tools/qualification/install/custody_witness.py
+++ b/tools/qualification/install/custody_witness.py
@@ -106,16 +106,18 @@ def fields(raw, types, depth=0):
 
 
 def validate(raw, schema, depth=0):
-    types = SCHEMAS[schema]
-    for number, values in fields(raw, types, depth).items():
-        spec = types.get(number)
+    parsed = fields(raw, SCHEMAS[schema], depth)
+    records = [parsed] if schema == 'record' else []
+    for number, values in parsed.items():
+        spec = SCHEMAS[schema].get(number)
         if isinstance(spec, str):
             for _, value in values:
                 if spec == 'utf8':
                     value.decode('utf-8')
                 else:
                     require(depth < 100, 'protobuf message recursion depth exhausted')
-                    validate(value, spec, depth + 1)
+                    records.extend(validate(value, spec, depth + 1))
+    return records
 
 
 def one(values, number, default=0):
@@ -136,21 +138,16 @@ def string_projection(values):
     return length_field(1, one(f, 1, b'')) if active == 1 else b''
 
 
-def log_rows(raw, batch):
-    index = 0
-    for _, resource in fields(raw, SCHEMAS['logs']).get(1, []):
-        for _, scope in fields(resource, SCHEMAS['resource_logs']).get(2, []):
-            for _, record in fields(scope, SCHEMAS['scope_logs']).get(2, []):
-                f = fields(record, SCHEMAS['record'])
-                projection = [b'\x59' + one(f, 11, b'\0' * 8), length_field(5, string_projection(f.get(5, [])))]
-                for _, attribute in f.get(6, []):
-                    kv = fields(attribute, SCHEMAS['kv'])
-                    projection.append(length_field(6, length_field(1, one(kv, 1, b''))
-                        + length_field(2, string_projection(kv.get(2, [])))))
-                row = oracle.decode_log_record(b''.join(projection))
-                yield dict(node='spindle-1', node_id=batch['node_id'].hex(), sequence=batch['sequence'],
-                    index=index, observed_ns=row['observed_time_unix_nano'], body=row['body'], attributes=row['attributes'])
-                index += 1
+def log_rows(records, batch):
+    for index, f in enumerate(records):
+        projection = [b'\x59' + one(f, 11, b'\0' * 8), length_field(5, string_projection(f.get(5, [])))]
+        for _, attribute in f.get(6, []):
+            kv = fields(attribute, SCHEMAS['kv'])
+            projection.append(length_field(6, length_field(1, one(kv, 1, b''))
+                + length_field(2, string_projection(kv.get(2, [])))))
+        row = oracle.decode_log_record(b''.join(projection))
+        yield dict(node='spindle-1', node_id=batch['node_id'].hex(), sequence=batch['sequence'],
+            index=index, observed_ns=row['observed_time_unix_nano'], body=row['body'], attributes=row['attributes'])
 
 
 def cursor(raw):
@@ -185,9 +182,8 @@ def envelope(raw):
             gap.decode('utf-8')
         cursors = [cursor(value) for _, value in f.get(7, [])]
         batch = dict(version=one(f, 1), node_id=one(f, 2, b''), generation=one(f, 3), sequence=one(f, 4))
-        for signal, schema in zip(signals, ('metrics', 'logs', 'traces')):
-            validate(signal, schema)
-        rows = list(log_rows(signals[1], batch))
+        decoded = [validate(signal, schema) for signal, schema in zip(signals, ('metrics', 'logs', 'traces'))]
+        rows = list(log_rows(decoded[1], batch))
         return batch, cursors, rows
     except (oracle.MalformedInput, UnicodeError, TypeError, struct.error) as error:
         raise ValueError(str(error)) from error
@@ -379,6 +375,18 @@ def conserve(before, after_install, after_drain, queries, candidate_ack_events):
         and not any(row['body'] in (queries['pending_body'], queries['unread_body']) for row in old), 'invalid old-row baseline')
     expected = old + [pending_row, unread_row]
     require(all(row in old + [pending_row] for source in pending for row in source['rows']), 'unexpected pending log row')
+    for path, current in after_drain['cursors'].items():
+        log, offset = after_drain['logs'][path], 0
+        for row in sorted((row for row in expected if row['attributes'].get('log.file.path') == path),
+                          key=lambda row: int(row['attributes']['log.file.offset.start'])):
+            attrs = row['attributes']
+            start, end = (int(attrs[key]) for key in ('log.file.offset.start', 'log.file.offset.end'))
+            require(start == offset and start < end <= current['offset']
+                and log['bytes'][start:end] == (row['body'] + '\n').encode()
+                and attrs['log.file.device'] == str(log['device']) and attrs['log.file.inode'] == str(log['inode']),
+                    'source prefix conservation')
+            offset = end
+        require(offset == current['offset'], 'source prefix conservation')
     keys = lambda rows: sorted((row['node_id'], row['sequence'], row['index']) for row in rows)
     encoded = lambda rows: sorted(json.dumps(row, sort_keys=True, allow_nan=False) for row in rows)
     require(len(set(keys(expected))) == len(expected), 'duplicate expected identity')