slop

server/src/services/roverLearningService/workers/dataset.py

@@ -0,0 +1,173 @@
+"""Read closed recordings with causal state/action alignment and bounded memory."""
+import bisect
+import collections
+import copy
+import json
+import math
+from pathlib import Path
+
+import av
+import numpy as np
+
+from policy import SLOTS, VALUES, HEIGHT, WIDTH, encode_command, sensor_vector
+
+
+def load_events(directory):
+    metadata = json.loads((directory / 'session.json').read_text())
+    if (metadata.get('version') != 1 or not metadata.get('endedAt') or metadata.get('error')
+            or metadata.get('droppedEvents') or metadata.get('reason') == 'interrupted'):
+        raise ValueError('Incomplete/interrupted recording or lost events')
+    if metadata.get('video', {}).get('origin') != 'unix':
+        raise ValueError('Unsupported video time origin')
+    video_result = metadata.get('video', {}).get('result') or {}
+    if video_result.get('signal') == 'SIGKILL' or video_result.get('code') not in (0, 255, None):
+        raise ValueError('Video recorder failed or was forcibly killed')
+    event_file = directory / 'events.ndjson'
+    if event_file.stat().st_size > 64 * 1024 * 1024:
+        raise ValueError('Event file exceeds 64 MiB read bound')
+    events = []
+    with event_file.open() as stream:
+        for line in stream:
+            event = json.loads(line)
+            if event['kind'] == 'captureGap':
+                raise ValueError('Recording contains a capture gap')
+            if not isinstance(event.get('ts'), (int, float)) or not math.isfinite(event['ts']):
+                raise ValueError('Invalid event timestamp')
+            events.append(event)
+    if not events or not any(e['kind'] == 'sessionEnd' for e in events):
+        raise ValueError('Missing completed event stream')
+    # File order can differ across rotation buffering. Monotonic time detects
+    # wall-clock steps; sorting Unix timestamps alone would hide those steps.
+    timed = sorted(events, key=lambda e: int(e['monotonicNs']))
+    origin = timed[0]
+    for event in timed:
+        drift = (event['ts'] - origin['ts']) - (int(event['monotonicNs']) - int(origin['monotonicNs'])) / 1e6
+        if abs(drift) > 250:
+            raise ValueError('Server wall clock shifted during recording')
+    return metadata, sorted(events, key=lambda e: e['ts'])
+
+
+def prepare(directory, schema):
+    metadata, events = load_events(directory)
+    requests = {}
+    accepted = set()
+    failed = set()
+    sensors, dispatches, boundaries = [], [], []
+    for event in events:
+        data = event['data']
+        if event['kind'] == 'sensor':
+            sensors.append((event['ts'], sensor_vector(data.get('sensors') or {})))
+        if event['kind'] in ('driver', 'turn:activeDriver', 'lock', 'private'):
+            boundaries.append(event['ts'])
+        if event['kind'] != 'command':
+            continue
+        if data.get('phase') == 'requested' and data.get('source') == 'client':
+            requests[data.get('requestId')] = data
+        if data.get('phase') == 'result' and data.get('result', {}).get('id'):
+            accepted.add(data['result']['id'])
+        if data.get('phase') == 'acknowledged' and data.get('acknowledgement', {}).get('error'):
+            failed.add(data.get('commandId'))
+        if data.get('phase') == 'dispatched':
+            dispatches.append(data)
+    if not sensors:
+        raise ValueError('No sensor observations')
+    candidate = copy.deepcopy(schema)
+    history, targets = [], []
+    for data in dispatches:
+        encoded = encode_command(data.get('command') or {}, candidate, extend=True)
+        if encoded is None:
+            continue
+        stamp = data['ts']
+        history.append((stamp, *encoded))
+        request = requests.get(data.get('requestId'))
+        if (data.get('source') == 'client' and request
+                and data.get('commandId') in accepted and data.get('commandId') not in failed):
+            target_time = request['ts']
+            if not 0 <= stamp - target_time <= 1000:
+                raise ValueError('Invalid request/dispatch timing')
+            targets.append((target_time, *encoded))
+        else:
+            # Server safety overrides belong in history, never demonstration
+            # targets. Exclude windows around them rather than teaching inaction.
+            boundaries.append(stamp)
+    if not targets:
+        raise ValueError('No accepted client operating commands')
+    schema[:] = candidate
+    return metadata, sensors, sorted(history, key=lambda x: x[0]), sorted(targets, key=lambda x: x[0]), sorted(boundaries)
+
+
+def windows(directory, schema, cancelled=lambda: False):
+    directory = Path(directory)
+    metadata, sensors, history, targets, boundaries = prepare(directory, schema)
+    sensor_times = [row[0] for row in sensors]
+    target_times = [row[0] for row in targets]
+    available = np.zeros(SLOTS, dtype=np.float32)
+    for _, slot, _ in history:
+        available[slot] = 1
+    command_values = np.zeros((SLOTS, VALUES), dtype=np.float32)
+    command_times = np.full(SLOTS, -np.inf)
+    cursor = 0
+    recent = collections.deque(maxlen=32)
+    last_sample, last_pts = -math.inf, -math.inf
+    with av.open(str(directory / 'video.mkv')) as container:
+        stream = container.streams.video[0]
+        stream.codec_context.thread_count = 1
+        for frame in container.decode(stream):
+            if cancelled():
+                return
+            if frame.pts is None:
+                raise ValueError('Video frame has no timestamp')
+            stamp = float(frame.pts * frame.time_base) * 1000
+            if not math.isfinite(stamp) or stamp < last_pts:
+                raise ValueError('Invalid/nonmonotonic video timestamps')
+            last_pts = stamp
+            if not metadata['startedAt'] - 2000 <= stamp <= metadata['endedAt'] + 2000:
+                raise ValueError('Video PTS is not aligned to session Unix time')
+            if stamp - last_sample < 99:
+                continue
+            last_sample = stamp
+            while cursor < len(history) and history[cursor][0] < stamp:
+                when, slot, values = history[cursor]
+                command_values[slot] = values
+                command_times[slot] = when
+                cursor += 1
+            sensor_index = bisect.bisect_right(sensor_times, stamp) - 1
+            if sensor_index < 0 or stamp - sensor_times[sensor_index] > 1000:
+                recent.clear()
+                continue
+            seen = np.isfinite(command_times).astype(np.float32)
+            ages = np.minimum((stamp - command_times) / 10000, 1).astype(np.float32)
+            state = np.concatenate((sensors[sensor_index][1], command_values.flatten(), ages, seen))
+            pixels = frame.to_ndarray(width=WIDTH, height=HEIGHT, format='rgb24').transpose(2, 0, 1).copy()
+            recent.append((stamp, pixels, state))
+            chosen = []
+            for offset in (900, 600, 300, 0):
+                options = [row for row in recent if row[0] <= stamp - offset + 0.01]
+                if not options or stamp - offset - options[-1][0] > 200:
+                    break
+                chosen.append(options[-1])
+            if len(chosen) != 4 or stamp + 100 > metadata['endedAt']:
+                continue
+            if bisect.bisect_right(boundaries, stamp + 100) > bisect.bisect_left(boundaries, chosen[0][0]):
+                continue
+            start = bisect.bisect_left(target_times, stamp)
+            end = bisect.bisect_left(target_times, stamp + 100)
+            actions = np.zeros(SLOTS, dtype=np.float32)
+            values = np.zeros((SLOTS, VALUES), dtype=np.float32)
+            ambiguous = False
+            families = set()
+            for _, slot, vector in targets[start:end]:
+                template = schema[slot]['template']
+                family = template['type']
+                if family == 'peripheral':
+                    peripheral = template.get('peripheral') or {}
+                    family += ':' + str(peripheral.get('id')) + ':' + str(peripheral.get('control'))
+                if ((actions[slot] and (not schema[slot]['fields'] or not np.array_equal(values[slot], vector)))
+                        or (not actions[slot] and family in families)):
+                    ambiguous = True
+                    break
+                families.add(family)
+                actions[slot], values[slot] = 1, vector
+            if ambiguous:
+                continue
+            yield np.stack([row[1] for row in chosen]), np.stack([row[2] for row in chosen]), actions, values, available

server/src/services/roverLearningService/workers/infer.py

@@ -0,0 +1,151 @@
+#!/usr/bin/env python3
+"""Live policy worker. JSON state on stdin, proposals on stdout; no rover writes."""
+import argparse
+import collections
+import hashlib
+import json
+import math
+from pathlib import Path
+import sys
+import threading
+import time
+
+import av
+import numpy as np
+import torch
+
+from policy import Policy, VERSION, SLOTS, VALUES, HEIGHT, WIDTH, encode_command, sensor_vector
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--model', required=True)
+    parser.add_argument('--url', required=True)
+    parser.add_argument('--threads', type=int, default=4)
+    parser.add_argument('--threshold', type=float, default=0.7)
+    args = parser.parse_args()
+    torch.set_num_threads(args.threads)
+    torch.set_num_interop_threads(1)
+    directory = Path(args.model)
+    metadata = json.loads((directory / 'model.json').read_text())
+    if metadata['specification']['version'] != VERSION:
+        raise ValueError('Model specification is incompatible')
+    if hashlib.sha256((directory / 'weights.pt').read_bytes()).hexdigest() != metadata['sha256']:
+        raise ValueError('Model weights checksum mismatch')
+    schema = metadata['specification']['commands']
+    model = Policy().eval()
+    model.load_state_dict(torch.load(directory / 'weights.pt', map_location='cpu', weights_only=True))
+    observations = collections.deque(maxlen=2048)
+    mutex = threading.Lock()
+    closed = threading.Event()
+
+    def read_input():
+        try:
+            for line in sys.stdin:
+                event = json.loads(line)
+                with mutex:
+                    observations.append(event)
+        finally:
+            closed.set()
+
+    threading.Thread(target=read_input, daemon=True).start()
+    values = np.zeros((SLOTS, VALUES), dtype=np.float32)
+    last_commands = np.full(SLOTS, -np.inf)
+    sensors, sensor_time = {}, 0
+    history = collections.deque(maxlen=32)
+    latched = set()
+    sampled = 0
+    with av.open(args.url, options={'rtsp_transport': 'tcp', 'timeout': '3000000',
+                                   'use_wallclock_as_timestamps': '1'}) as video:
+        stream = video.streams.video[0]
+        stream.codec_context.thread_count = 1
+        for frame in video.decode(stream):
+            if closed.is_set():
+                return
+            if frame.pts is None:
+                continue
+            stamp = float(frame.pts * frame.time_base) * 1000
+            now = time.time() * 1000
+            if not math.isfinite(stamp) or abs(now - stamp) > 2000:
+                raise ValueError('Video time is stale or not aligned to server time')
+            if stamp - sampled < 99:
+                continue
+            sampled = stamp
+            with mutex:
+                while observations and observations[0]['ts'] < stamp:
+                    event = observations.popleft()
+                    if event['kind'] == 'sensor':
+                        sensors, sensor_time = event['data'], event['ts']
+                    elif event['kind'] == 'command':
+                        try:
+                            encoded = encode_command(event['data'], schema)
+                            if encoded:
+                                slot, vector = encoded
+                                values[slot], last_commands[slot] = vector, event['ts']
+                        except ValueError:
+                            pass  # Unseen manual controls do not stop autonomy.
+            if stamp - sensor_time > 1000:
+                history.clear()
+                continue
+            state = np.concatenate((sensor_vector(sensors), values.flatten(),
+                                    np.minimum((stamp - last_commands) / 10000, 1).astype(np.float32),
+                                    np.isfinite(last_commands).astype(np.float32)))
+            pixels = frame.to_ndarray(width=WIDTH, height=HEIGHT, format='rgb24').transpose(2, 0, 1).copy()
+            history.append((stamp, pixels, state))
+            selected = []
+            for offset in (900, 600, 300, 0):
+                candidates = [entry for entry in history if entry[0] <= stamp - offset + .01]
+                if not candidates or stamp - offset - candidates[-1][0] > 200:
+                    break
+                selected.append(candidates[-1])
+            if len(selected) != 4:
+                continue
+            before = time.perf_counter()
+            with torch.inference_mode():
+                logits, numeric = model(torch.from_numpy(np.stack([row[1] for row in selected])).unsqueeze(0).float() / 255,
+                                        torch.from_numpy(np.stack([row[2] for row in selected])).unsqueeze(0))
+                probabilities = logits[0].sigmoid().tolist()
+                predictions = numeric[0].tolist()
+            if not all(math.isfinite(value) for value in probabilities):
+                raise ValueError('Nonfinite model output')
+            families = {}
+            active = set()
+            for slot, descriptor in enumerate(schema):
+                if probabilities[slot] < args.threshold:
+                    continue
+                active.add(slot)
+                command = json.loads(json.dumps(descriptor['template']))
+                family = command['type']
+                if family == 'peripheral':
+                    family += ':' + str(command['peripheral']['id']) + ':' + str(command['peripheral']['control'])
+                for index, field in enumerate(descriptor['fields']):
+                    number = predictions[slot][index]
+                    if not math.isfinite(number):
+                        raise ValueError('Nonfinite control value')
+                    number = math.tan(max(-.98, min(.98, number)) * math.pi / 2) * field['scale']
+                    target = command
+                    for key in field['path'][:-1]:
+                        target = target[key]
+                    target[field['path'][-1]] = round(number) if field['integer'] else number
+                discrete = command['type'] in ('headlight', 'laser', 'horn', 'song', 'raw')
+                if family not in families or probabilities[slot] > families[family][0]:
+                    families[family] = (probabilities[slot], command, slot, discrete)
+            latched.intersection_update(active)
+            # Choose the winning action before latching, so a held toggle cannot
+            # expose a competing lower-confidence toggle on the following tick.
+            commands = [command for _, command, slot, discrete in families.values()
+                        if not discrete or slot not in latched]
+            for _, _, slot, discrete in families.values():
+                if discrete:
+                    latched.add(slot)
+            print(json.dumps({'kind': 'prediction', 'frameAt': stamp, 'sensorAt': sensor_time,
+                              'latencyMs': (time.perf_counter() - before) * 1000,
+                              'commands': commands}, allow_nan=False), flush=True)
+
+
+if __name__ == '__main__':
+    try:
+        main()
+    except Exception as error:
+        print(json.dumps({'kind': 'error', 'message': str(error)}), flush=True)
+        raise SystemExit(1)

server/src/services/roverLearningService/workers/policy.py

@@ -0,0 +1,131 @@
+"""Versioned, fixed-size vision/control policy and command encoding.
+
+Slots describe command shapes, not arbitrary executable instructions. Numeric
+leaves are regressed; string/bool/list values remain exact template constants.
+No command is sent by this module. Execution must use server authorization.
+"""
+import copy
+import json
+import math
+
+import numpy as np
+import torch
+from torch import nn
+
+VERSION = 1
+SLOTS = 64
+VALUES = 8
+FRAMES = 4
+HEIGHT, WIDTH = 120, 160
+OPERATING_TYPES = {"drive", "motors", "servo", "headlight", "laser", "horn", "peripheral", "song", "raw"}
+SENSORS = [
+    ("bumpsAndWheelDrops.bumpLeft", 1), ("bumpsAndWheelDrops.bumpRight", 1),
+    ("bumpsAndWheelDrops.wheelDropLeft", 1), ("bumpsAndWheelDrops.wheelDropRight", 1),
+    ("cliffLeft", 1), ("cliffFrontLeft", 1), ("cliffFrontRight", 1), ("cliffRight", 1),
+    ("wall", 1), ("virtualWall", 1), ("wallSignal", 4096),
+    ("voltageMv", 16000), ("currentMa", 3000), ("batteryChargeMah", 4000),
+    ("batteryCapacityMah", 4000), ("batteryTemperatureC", 50),
+    ("chargingState.code", 5), ("chargingSources.homeBase", 1), ("oiMode.code", 4),
+    ("wheelSpeedsMmPerSecond.left", 500), ("wheelSpeedsMmPerSecond.right", 500),
+    ("wheelLeftCurrentMa", 2000), ("wheelRightCurrentMa", 2000),
+    ("mainBrushCurrentMa", 2000), ("sideBrushCurrentMa", 2000),
+    ("lightBumpLeftSignal", 4096), ("lightBumpRightSignal", 4096),
+    ("cliffLeftSignal", 4096), ("cliffRightSignal", 4096),
+    ("distanceMm", 500), ("angleDeg", 180), ("stasis", 1),
+]
+STATE_SIZE = len(SENSORS) * 2 + SLOTS * (VALUES + 2)
+
+
+def normalize(value, scale):
+    return math.atan(float(value) / scale) * 2 / math.pi
+
+
+def sensor_vector(sensors):
+    values, present = [], []
+    for name, scale in SENSORS:
+        value = sensors
+        for key in name.split('.'):
+            value = value.get(key) if isinstance(value, dict) else None
+        known = isinstance(value, (int, float)) and math.isfinite(value)
+        values.append(normalize(value, scale) if known else 0)
+        present.append(float(known))
+    return np.array(values + present, dtype=np.float32)
+
+
+def describe_command(command):
+    if command.get('type') not in OPERATING_TYPES:
+        return None
+    if len(json.dumps(command)) > 4096:
+        raise ValueError('Operating command exceeds 4096-byte schema limit')
+    values, fields = [], []
+
+    def visit(value, path=()):
+        if isinstance(value, dict):
+            return {key: visit(item, (*path, key)) for key, item in sorted(value.items())}
+        if isinstance(value, (int, float)) and not isinstance(value, bool):
+            if not math.isfinite(value) or len(values) >= VALUES:
+                raise ValueError('Nonfinite command or too many numeric control fields')
+            scale = {'drive': 500, 'motors': 255, 'servo': 180, 'peripheral': 255,
+                     'horn': 1000, 'song': 127}.get(command['type'], 1)
+            fields.append({'path': list(path), 'scale': scale, 'integer': isinstance(value, int)})
+            values.append(normalize(value, scale))
+            return {'$number': len(values) - 1}
+        # Arrays (e.g. songs) are discrete exact demonstrations, not variable
+        # length regression outputs. Never reinterpret raw OI bytes as numbers.
+        return copy.deepcopy(value)
+
+    template = visit(command)
+    return {'template': template, 'fields': fields}, values
+
+
+def encode_command(command, schema, extend=False):
+    described = describe_command(command)
+    if described is None:
+        return None
+    descriptor, values = described
+    try:
+        slot = schema.index(descriptor)
+    except ValueError:
+        if not extend:
+            raise ValueError('Command shape missing from model schema')
+        if len(schema) >= SLOTS:
+            raise ValueError('64 command shapes exhausted; recording skipped, not silently truncated')
+        slot = len(schema)
+        schema.append(descriptor)
+    vector = np.zeros(VALUES, dtype=np.float32)
+    vector[:len(values)] = values
+    return slot, vector
+
+
+class Policy(nn.Module):
+    def __init__(self):
+        super().__init__()
+        self.vision = nn.Sequential(
+            nn.Conv2d(3, 24, 5, 2, 2), nn.ReLU(),
+            nn.Conv2d(24, 48, 3, 2, 1), nn.ReLU(),
+            nn.Conv2d(48, 96, 3, 2, 1), nn.ReLU(),
+            nn.Conv2d(96, 128, 3, 2, 1), nn.ReLU(),
+            nn.AdaptiveAvgPool2d((3, 4)), nn.Flatten(),
+            nn.Linear(128 * 3 * 4, 128), nn.ReLU(),
+        )
+        self.history = nn.GRU(128 + STATE_SIZE, 128, batch_first=True)
+        self.actions = nn.Linear(128, SLOTS)
+        self.values = nn.Linear(128, SLOTS * VALUES)
+
+    def forward(self, images, state):
+        batch, frames = images.shape[:2]
+        encoded = self.vision(images.reshape(-1, 3, HEIGHT, WIDTH)).reshape(batch, frames, -1)
+        history, _ = self.history(torch.cat((encoded, state), dim=-1))
+        last = history[:, -1]
+        return self.actions(last), self.values(last).reshape(batch, SLOTS, VALUES).tanh()
+
+
+def specification(schema):
+    return {'version': VERSION, 'architecture': 'spatial-cnn-gru-v1',
+            'frames': FRAMES, 'height': HEIGHT, 'width': WIDTH,
+            'historyStrideMs': 300, 'decisionIntervalMs': 100,
+            'stateSize': STATE_SIZE, 'sensorFields': SENSORS,
+            'normalization': '2/pi * atan(value/scale); missing sensors have mask=0',
+            'maxCommandShapes': SLOTS, 'numericFieldsPerShape': VALUES,
+            'commands': schema, 'actionSemantics': 'Events in the next 100 ms; zero means no new command',
+            'timing': 'Server-receipt alignment; browser display latency unmeasured'}

server/src/services/roverLearningService/workers/requirements.txt

@@ -0,0 +1,4 @@
+--extra-index-url https://download.pytorch.org/whl/cpu
+torch==2.14.0+cpu
+numpy==2.5.3
+av==18.1.0

server/src/services/roverLearningService/workers/train.py

@@ -0,0 +1,214 @@
+#!/usr/bin/env python3
+"""One finite CPU training job; Node owns scheduling, leases, and shutdown.
+
+Usage: python train.py --job /path/to/job.json
+Dependencies are packaged at image build time, never installed at runtime.
+"""
+import argparse
+import copy
+import hashlib
+import json
+import os
+from pathlib import Path
+import random
+import signal
+import time
+import uuid
+
+
+def emit(kind, **data):
+    print(json.dumps({'kind': kind, **data}, allow_nan=False), flush=True)
+
+
+def atomic_torch_save(torch, value, path):
+    temporary = path.with_suffix('.tmp')
+    with temporary.open('wb') as stream:
+        torch.save(value, stream)
+        stream.flush()
+        os.fsync(stream.fileno())
+    temporary.replace(path)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--job')
+    parser.add_argument('--check-dependencies', action='store_true', help='Describe the installed production policy and exit without creating storage')
+    args = parser.parse_args()
+    # --help works even before Python dependencies are installed.
+    import numpy as np
+    import av
+    import torch
+    from torch.nn import functional as functional
+    from dataset import windows
+    from policy import Policy, VERSION, SLOTS, VALUES, specification
+
+    if args.check_dependencies:
+        emit('dependencies', torch=str(torch.__version__), numpy=str(np.__version__), av=str(av.__version__),
+             parameters=sum(p.numel() for p in Policy().parameters()), specification=specification([]))
+        return
+    if not args.job:
+        parser.error('--job is required unless --check-dependencies is used')
+
+    job = json.loads(Path(args.job).read_text())
+    root = Path(job['root'])
+    config = job['config']
+    torch.set_num_threads(config['threads'])
+    torch.set_num_interop_threads(1)
+    # Training yields CPU priority to the Node/media processes. Thread count is
+    # an additional bound, not a claim to reserve specific physical cores.
+    os.nice(10)
+    stopped = False
+
+    def stop(_signal, _frame):
+        nonlocal stopped
+        stopped = True
+
+    signal.signal(signal.SIGTERM, stop)
+    signal.signal(signal.SIGINT, stop)
+    started = time.monotonic()
+    cancelled = lambda: stopped or time.monotonic() - started >= config['maxJobSeconds']
+    model = Policy().cpu()
+    optimizer = torch.optim.AdamW(model.parameters(), lr=0.0001)
+    resume_path = root / 'training' / 'resume.pt'
+    schema, steps, examples, last_published = [], 0, 0, 0
+    trained_rovers = set()
+    control_profiles = []
+    if resume_path.exists():
+        saved = torch.load(resume_path, map_location='cpu', weights_only=True)
+        if saved['version'] != VERSION:
+            raise ValueError('Resume schema incompatible; refusing to reinterpret existing weights')
+        model.load_state_dict(saved['weights'])
+        optimizer.load_state_dict(saved['optimizer'])
+        schema = saved['schema']
+        steps, examples = saved['steps'], saved['examples']
+        last_published = saved['lastPublished']
+        trained_rovers = set(saved['trainedRovers'])
+        control_profiles = saved.get('controlProfiles', [])
+    initial_steps = steps
+    pool, accepted, rejected = [], [], {}
+    sessions = list(job['sessions'])
+    random.shuffle(sessions)
+    per_session = max(1, config['maxSamples'] // max(1, len(sessions)))
+    for session_id in sessions:
+        if cancelled():
+            break
+        candidate_schema = copy.deepcopy(schema)
+        samples, seen = [], 0
+        try:
+            directory = root / 'recordings' / session_id
+            for sample in windows(directory, candidate_schema, cancelled):
+                seen += 1
+                # Uniform reservoir across the whole recording, not just the
+                # opening seconds. Frames remain uint8 until a batch is used.
+                if len(samples) < per_session:
+                    samples.append(sample)
+                else:
+                    index = random.randrange(seen)
+                    if index < per_session:
+                        samples[index] = sample
+            if cancelled():
+                break  # Do not call a partially decoded recording validated.
+            if not samples:
+                raise ValueError('No complete, aligned observation/action windows')
+            schema = candidate_schema
+            pool.extend(samples)
+            accepted.append(session_id)
+            metadata = json.loads((directory / 'session.json').read_text())
+            trained_rovers.add(str(metadata['roverId']))
+            profile = metadata.get('controlProfile')
+            if profile is not None and profile not in control_profiles:
+                control_profiles.append(profile)
+            emit('dataset', sessionId=session_id, windows=seen, sampled=len(samples))
+        except (ValueError, KeyError, TypeError, OSError, EOFError, IndexError, OverflowError, av.error.FFmpegError) as error:
+            rejected[session_id] = str(error)[:500]
+            emit('skipped', sessionId=session_id, reason=rejected[session_id])
+    if len(pool) < config['minimumSamples'] or not any(sample[2].any() for sample in pool) or cancelled():
+        emit('complete', trained=False, accepted=[], rejected=rejected, samples=len(pool),
+             reason='Stopped/time limit' if cancelled() else 'Waiting for more usable samples')
+        return
+
+    numeric_mask = torch.zeros(SLOTS, VALUES)
+    for slot, descriptor in enumerate(schema):
+        numeric_mask[slot, :len(descriptor['fields'])] = 1
+    # Balance sparse events without assuming that every time step should fire
+    # every control. This is training loss, not a driving-quality evaluation.
+    positives = np.sum([sample[2] for sample in pool], axis=0)
+    opportunities = np.sum([sample[4] for sample in pool], axis=0)
+    positive_weight = torch.from_numpy(np.clip((opportunities - positives) / np.maximum(positives, 1), 1, 20)).float()
+    losses = []
+    model.train()
+    for _ in range(config['passesPerJob']):
+        random.shuffle(pool)
+        for offset in range(0, len(pool), config['batchSize']):
+            if cancelled() or steps - initial_steps >= config['maxStepsPerJob']:
+                break
+            batch = pool[offset:offset + config['batchSize']]
+            images, states, targets, values, available = [torch.from_numpy(np.stack(items)) for items in zip(*batch)]
+            images = images.float().div_(255)
+            optimizer.zero_grad(set_to_none=True)
+            action_logits, predictions = model(images, states)
+            event_loss = functional.binary_cross_entropy_with_logits(action_logits, targets,
+                                                                       pos_weight=positive_weight, reduction='none')
+            event_loss = (event_loss * available).sum() / available.sum().clamp_min(1)
+            mask = numeric_mask.unsqueeze(0) * targets.unsqueeze(-1)
+            value_loss = ((predictions - values).square() * mask).sum() / mask.sum().clamp_min(1)
+            loss = event_loss + value_loss
+            if not torch.isfinite(loss):
+                raise ValueError('Nonfinite training loss; checkpoint not saved')
+            loss.backward()
+            torch.nn.utils.clip_grad_norm_(model.parameters(), 1)
+            optimizer.step()
+            steps += 1
+            examples += len(batch)
+            losses.append(loss.item())
+            if steps % 10 == 0:
+                emit('progress', steps=steps, examples=examples, loss=loss.item())
+        if cancelled() or steps - initial_steps >= config['maxStepsPerJob']:
+            break
+    if steps == initial_steps:
+        emit('complete', trained=False, accepted=[], rejected=rejected, reason='No training steps completed')
+        return
+
+    published = None
+    if steps - last_published >= config['checkpointEverySteps']:
+        adjectives = ['amber', 'curious', 'gentle', 'quiet', 'silver', 'bright', 'merry', 'sleepy']
+        animals = ['otter', 'finch', 'badger', 'robin', 'fox', 'heron', 'marten', 'wren']
+        model_id = str(uuid.uuid4())
+        name = f'{random.choice(adjectives)}-{random.choice(animals)}-{model_id[:8]}'
+        staging = root / 'models' / f'.{model_id}.pending'
+        staging.mkdir()
+        weights_path = staging / 'weights.pt'
+        atomic_torch_save(torch, model.state_dict(), weights_path)
+        published = {
+            'id': model_id, 'name': name, 'createdAt': int(time.time() * 1000),
+            'trainingSteps': steps, 'examplesProcessed': examples,
+            'parameters': sum(p.numel() for p in model.parameters()),
+            'weightBytes': weights_path.stat().st_size,
+            'sha256': hashlib.sha256(weights_path.read_bytes()).hexdigest(),
+            'specification': specification(schema), 'trainedRovers': sorted(trained_rovers),
+            'controlProfiles': control_profiles,
+            'sourceSessions': accepted, 'trainingLoss': sum(losses) / len(losses),
+            'evaluation': 'unevaluated', 'dtype': 'float32', 'torchVersion': str(torch.__version__),
+        }
+        (staging / 'model.json').write_text(json.dumps(published, indent=2, allow_nan=False))
+        # A catalog reader sees both weights and metadata, or neither. Existing
+        # versions are never overwritten or deleted by the trainer.
+        staging.rename(root / 'models' / model_id)
+        last_published = steps
+    atomic_torch_save(torch, {
+        'version': VERSION, 'weights': model.state_dict(), 'optimizer': optimizer.state_dict(),
+        'schema': schema, 'steps': steps, 'examples': examples,
+        'lastPublished': last_published, 'trainedRovers': sorted(trained_rovers),
+        'controlProfiles': control_profiles,
+    }, resume_path)
+    emit('complete', trained=True, accepted=accepted, rejected=rejected,
+         steps=steps, examples=examples, loss=sum(losses) / len(losses),
+         published=published, elapsedSeconds=time.monotonic() - started)
+
+
+if __name__ == '__main__':
+    try:
+        main()
+    except Exception as error:
+        emit('error', message=str(error))
+        raise SystemExit(1)