slop

server/src/services/roverLearningService/configuration.js

@@ -11,7 +11,8 @@ module.exports = {
     training: {
       enabled: true, python: '/opt/rover-learning/bin/python', threads: 8,
       batchSize: 16, maxSamples: 1024, minimumSamples: 128, passesPerJob: 2,
-      roundsPerSession: 3, maxStepsPerJob: 200, checkpointEverySteps: 100,
+      maxStepsPerJob: 200, checkpointEverySteps: 1000,
+      minimumDrivingMinutes: 30, newDrivingMinutesPerModel: 10,
       intervalSeconds: 60, maxJobSeconds: 600,
     },
   },
@@ -35,11 +36,12 @@ module.exports = {
       maxSamples: integer({ minimum: 128, maximum: 2048, description: 'Maximum decoded windows retained in RAM per job. Each contains four 160x120 RGB frames.' }),
       minimumSamples: integer({ minimum: 1, maximum: 128, description: 'Usable sampled windows required before training.' }),
       passesPerJob: integer({ minimum: 1, maximum: 10, description: 'Shuffled passes over the sampled windows per job.' }),
-      roundsPerSession: integer({ minimum: 1, maximum: 20, description: 'Maximum successful sampled training visits per retained recording; pauses repeated training on unchanged data.' }),
+      minimumDrivingMinutes: number({ minimum: 1, maximum: 100000, description: 'Minimum distinct usable training footage in minutes before publishing a model; repeated passes do not count.' }),
+      newDrivingMinutesPerModel: number({ minimum: 1, maximum: 100000, description: 'Additional distinct usable footage required between published models.' }),
       maxStepsPerJob: integer({ minimum: 1, maximum: 10000, description: 'Optimizer step cap per finite training job.' }),
       checkpointEverySteps: integer({ minimum: 1, maximum: 100000, description: 'Accumulated training steps between permanent named model publications.' }),
       intervalSeconds: integer({ minimum: 30, maximum: 86400, description: 'Minimum seconds between completed training jobs.' }),
       maxJobSeconds: integer({ minimum: 30, maximum: 3600, description: 'Wall-clock budget including video decoding; slow/stuck workers are stopped.' }),
-    }, { description: 'Bounded CPU training, sampling, and permanent checkpoint publication.', required: ['enabled', 'python', 'threads', 'batchSize', 'maxSamples', 'minimumSamples', 'passesPerJob', 'roundsPerSession', 'maxStepsPerJob', 'checkpointEverySteps', 'intervalSeconds', 'maxJobSeconds'] }),
+    }, { description: 'Bounded CPU training, sampling, and permanent checkpoint publication.', required: ['enabled', 'python', 'threads', 'batchSize', 'maxSamples', 'minimumSamples', 'passesPerJob', 'minimumDrivingMinutes', 'newDrivingMinutesPerModel', 'maxStepsPerJob', 'checkpointEverySteps', 'intervalSeconds', 'maxJobSeconds'] }),
   }, { description: 'Optional rover recording and learning service.', required: ['enabled', 'recording', 'training', 'driving'] }),
 };

server/src/services/roverLearningService/driving.js

@@ -42,12 +42,13 @@ function createDriving({ root, config, onChange, activity }) {
     }
     if (command.type === 'servo') {
       const servo = record.meta?.cameraServo;
-      if (!servo?.enabled) return;
+      if (!servo?.enabled) { session.submissions.push({ command, result: 'Servo unavailable' }); return; }
       command.servo.angle = clamp(command.servo?.angle, servo.minAngle ?? 0, servo.maxAngle ?? 180);
     }
     const { type, ...data } = command;
     recordCommandRequest(session.socket, { roverId: session.roverId, type, data }, (result) => {
       session.lastResult = result.error || result.reason || 'issued';
+      session.submissions.push({ command, result: session.lastResult });
       if (result.id) session.lastAction = { at: Date.now(), command };
     }, (request, reply) => submitCommand(session.socket, request, reply), 'model');
   }
@@ -118,7 +119,11 @@ function createDriving({ root, config, onChange, activity }) {
           session.frameAt = message.frameAt;
           session.sensorAt = message.sensorAt;
           session.latencyMs = message.latencyMs;
+          session.proposals = Array.isArray(message.proposals) ? message.proposals.slice(0, 64) : [];
+          session.threshold = message.threshold;
+          session.submissions = [];
           if (!Array.isArray(message.commands) || message.commands.length > 64) throw new Error('Invalid action batch');
+          session.predictionStatus = message.commands.length ? 'Proposing actions' : 'Predicting no new action';
           // Manual commands intentionally do not stop or pause this controller.
           for (const command of message.commands) transmit(session, command);
         } catch (error) { stop(roverId, error.message); break; }

server/src/services/roverLearningService/recording.js

@@ -15,7 +15,7 @@ async function saveMetadata(directory, metadata) {
   await fsp.rename(`${filename}.tmp`, filename);
 }
 
-async function createRecording({ root, roverId, snapshot, logger }) {
+async function createRecording({ root, roverId, snapshot, logger, docked = false }) {
   const id = `${Date.now()}-${randomUUID()}`;
   const directory = path.join(root, id);
   await fsp.mkdir(directory);
@@ -24,10 +24,15 @@ async function createRecording({ root, roverId, snapshot, logger }) {
     controlProfile: controlProfile(snapshot.meta),
     timing: 'Video PTS uses server receiver Unix seconds. Events use Unix milliseconds and process monotonic nanoseconds. Browser display delay is not measured.',
     video: { file: 'video.mkv', codec: 'source-copy', timestampUnit: 'seconds', origin: 'unix' },
-    events: 'events.ndjson', droppedEvents: 0,
+    events: 'events.ndjson', droppedEvents: 0, bufferOnly: docked,
+    trainingStartedAt: docked ? null : Date.now(),
   };
   await saveMetadata(directory, metadata);
   const output = fs.createWriteStream(path.join(directory, 'events.ndjson'), { flags: 'wx' });
+  let buffered = [];
+  let bufferedBytes = 0;
+  let dockedAt = null;
+  let pruning = Promise.resolve();
   let failure = null;
   let stopping = false;
   let stopPromise;
@@ -49,6 +54,19 @@ async function createRecording({ root, roverId, snapshot, logger }) {
       const line = JSON.stringify({
         ts: Date.now(), monotonicNs: process.hrtime.bigint().toString(), ...timestamp, kind, data,
       }) + '\n';
+      if (metadata.bufferOnly) {
+        const now = Date.now();
+        buffered.push({ at: now, line });
+        bufferedBytes += Buffer.byteLength(line);
+        while (buffered.length && buffered[0].at < now - 5000) {
+          bufferedBytes -= Buffer.byteLength(buffered.shift().line);
+        }
+        while (bufferedBytes > MAX_BUFFER_BYTES && buffered.length) {
+          metadata.droppedEvents += 1;
+          bufferedBytes -= Buffer.byteLength(buffered.shift().line);
+        }
+        return true;
+      }
       // Never wait for disk from a control/sensor callback. Explicit loss counts
       // prevent an overloaded recorder from pretending its history is complete.
       if (output.writableLength + Buffer.byteLength(line) > MAX_BUFFER_BYTES) {
@@ -73,7 +91,10 @@ async function createRecording({ root, roverId, snapshot, logger }) {
     '-use_wallclock_as_timestamps', '1', '-copyts',
     '-i', `rtsp://127.0.0.1:8554/${encodeURIComponent(roverId)}`,
     '-map', '0:v:0', '-an', '-c:v', 'copy', '-avoid_negative_ts', 'disabled',
-    '-f', 'matroska', path.join(directory, 'video.mkv'),
+    // Keyframe-aligned chunks keep a disposable dock buffer without repeatedly
+    // reconnecting to RTSP. Extra edge packets are filtered by trainingStartedAt.
+    '-f', 'segment', '-segment_time', '2', '-segment_format', 'matroska',
+    '-reset_timestamps', '0', path.join(directory, 'video-%09d.mkv'),
   ], { stdio: ['ignore', 'ignore', 'pipe'] });
   child.stderr.on('data', (chunk) => { stderr = (stderr + chunk.toString()).slice(-4096); });
   child.on('error', (error) => { failure = error.message; });
@@ -84,6 +105,35 @@ async function createRecording({ root, roverId, snapshot, logger }) {
     resolveVideo();
   });
 
+  function updateDock(isDocked) {
+    if (stopping) return;
+    if (!isDocked) {
+      dockedAt = null;
+      if (metadata.bufferOnly) {
+        metadata.bufferOnly = false;
+        metadata.trainingStartedAt = Date.now() - 5000;
+        for (const item of buffered) output.write(item.line);
+        buffered = [];
+        bufferedBytes = 0;
+      }
+    } else if (!metadata.bufferOnly && dockedAt == null) dockedAt = Date.now();
+  }
+
+  function pruneBuffer() {
+    pruning = pruning.then(async () => {
+      if (!metadata.bufferOnly || stopping) return;
+      const files = (await fsp.readdir(directory)).filter((name) => /^video-\d+\.mkv$/.test(name)).sort();
+      // Never unlink the open segment; retain five seconds plus the preceding
+      // keyframe segment so the decoder can recover the full lead-in.
+      for (const name of files.slice(0, -2)) {
+        const filename = path.join(directory, name);
+        const stat = await fsp.stat(filename);
+        if (!metadata.bufferOnly || stopping) return;
+        if (stat.mtimeMs < Date.now() - 5000) await fsp.unlink(filename);
+      }
+    }).catch((error) => { failure = error.message; child.kill('SIGTERM'); });
+  }
+
   async function stop(reason) {
     if (stopPromise) return stopPromise;
     stopPromise = (async () => {
@@ -93,12 +143,18 @@ async function createRecording({ root, roverId, snapshot, logger }) {
       // A stuck RTSP read must not leave a recorder behind after disable.
       const killTimer = setTimeout(() => { if (!childClosed) child.kill('SIGKILL'); }, 2000);
       await videoClosed;
+      await pruning;
       clearTimeout(killTimer);
       await new Promise((resolve) => {
         if (output.closed) return resolve();
         output.once('close', resolve);
         output.end();
       });
+      if (metadata.bufferOnly) {
+        await fsp.rm(directory, { recursive: true, force: true });
+        return;
+      }
+      metadata.video.files = (await fsp.readdir(directory)).filter((name) => /^video-\d+\.mkv$/.test(name)).sort();
       metadata.endedAt = Date.now();
       metadata.reason = reason;
       metadata.error = failure;
@@ -110,11 +166,14 @@ async function createRecording({ root, roverId, snapshot, logger }) {
   }
 
   return {
-    id, directory, startedAt: metadata.startedAt, append, stop,
+    id, directory, startedAt: metadata.startedAt, append, stop, updateDock, pruneBuffer,
+    get buffered() { return metadata.bufferOnly; },
+    get dockTailComplete() { return dockedAt != null && Date.now() - dockedAt >= 5000; },
+    get trainingStartedAt() { return metadata.trainingStartedAt; },
     get failed() { return Boolean(failure); },
     get stopping() { return stopping; },
     getStatus: () => ({ id, roverId, startedAt: metadata.startedAt, droppedEvents: metadata.droppedEvents,
-      video: childClosed ? 'stopped' : 'recorder running', error: failure, stopping }),
+      video: childClosed ? 'stopped' : metadata.bufferOnly ? 'Dock lead-in buffer' : 'Recording human driving', error: failure, stopping }),
     // Process exit is synchronous; normal disable uses stop() and awaits close.
     kill: () => { if (!childClosed) child.kill('SIGKILL'); },
   };

server/src/services/roverLearningService/runtime.js

@@ -28,9 +28,10 @@ function createRuntime({ config, logger }) {
   logger = Object.fromEntries(['info', 'warn', 'error'].map((level) => [level, (message, details) => {
     baseLogger[level](message, details); activity(message, details);
   }]));
-  const trainer = createTrainer({ root, config: config.training, logger });
+  const trainer = createTrainer({ root, config: config.training, actionThreshold: config.driving.actionThreshold, logger });
   const driving = createDriving({ root, config, onChange: refreshEligibility, activity });
   let gateway;
+  let dockTimer;
   let diskState = { bytes: null, free: null };
   const sessions = new Map();
   const unsubscribers = [];
@@ -134,8 +135,8 @@ function createRuntime({ config, logger }) {
     pendingBytes = pending.reduce((sum, event) => sum + event.bytes, 0);
     for (const [roverId, session] of sessions) {
       const record = roverManager.rovers.get(roverId);
-      if (!hasHumanController(roverId) || session.stopping || session.failed || Date.now() - session.startedAt >= config.recording.sessionSeconds * 1000) {
-        await closeSession(roverId, !record?.ws ? 'offline' : !hasHumanController(roverId) ? 'no-human-controller' : session.failed ? 'recorder-failed' : 'rotation');
+      if (!hasHumanController(roverId) || session.stopping || session.failed || session.dockTailComplete || (!session.buffered && Date.now() - session.trainingStartedAt >= config.recording.sessionSeconds * 1000)) {
+        await closeSession(roverId, !record?.ws ? 'offline' : !hasHumanController(roverId) ? 'no-human-controller' : session.failed ? 'recorder-failed' : session.dockTailComplete ? 'docked' : 'rotation');
       }
     }
     if (!running) return;
@@ -160,7 +161,7 @@ function createRuntime({ config, logger }) {
       if (!running) break;
       const roverId = String(key);
       if (!hasHumanController(roverId) || sessions.has(roverId)) continue;
-      const session = await createRecording({ root: recordingsRoot, roverId, snapshot: snapshot(record), logger });
+      const session = await createRecording({ root: recordingsRoot, roverId, snapshot: snapshot(record), logger, docked: Boolean(record.lastSensor?.decoded?.chargingSources?.homeBase) });
       sessions.set(roverId, session);
       // Filesystem awaits can span a disconnect or turn change.
       if (!running || !hasHumanController(roverId)) {
@@ -223,6 +224,13 @@ function createRuntime({ config, logger }) {
     await trainer.start();
     running = true;
     driving.attach();
+    dockTimer = setInterval(() => {
+      for (const [id, session] of sessions) {
+        session.updateDock(Boolean(roverManager.rovers.get(id)?.lastSensor?.decoded?.chargingSources?.homeBase));
+        session.pruneBuffer();
+        if (session.dockTailComplete) schedule();
+      }
+    }, 250);
     gateway = createGateway({ root, driving, activity, getSystemState: () => ({
       recording: { paused: diskPaused, bytes: diskState.bytes, freeBytes: diskState.free,
         pendingBytes, droppedPending, sessions: [...sessions.values()].map((session) => session.getStatus()) },
@@ -277,6 +285,7 @@ function createRuntime({ config, logger }) {
     running = false;
     gateway?.close();
     clearTimeout(timer);
+    clearInterval(dockTimer);
     for (const unsubscribe of unsubscribers.splice(0)) unsubscribe();
     stopPromise = (async () => {
       const trainingStop = trainer.stop();

server/src/services/roverLearningService/socketGateway.js

@@ -1,4 +1,4 @@
-// One complete card snapshot per subscriber; no telemetry-frequency UI events.
+// Complete card snapshots; live diagnostics opt into 10 Hz, ordinary cards use 1 Hz.
 const io = require('../../globals/io');
 const roverManager = require('../roverManager');
 const { isVerified } = require('../verificationService');
@@ -13,6 +13,7 @@ function createGateway({ root, driving, getSystemState, activity }) {
   let lastCatalog = 0;
   let closed = false;
   let refreshing = false;
+  let lastOrdinary = 0;
 
   function stateFor(socket, roverId) {
     if (!isVerified(socket)) return { available: false };
@@ -30,7 +31,7 @@ function createGateway({ root, driving, getSystemState, activity }) {
         .map((model) => ({ id: model.id, name: model.name, createdAt: model.createdAt,
           trainingSteps: model.trainingSteps, examplesProcessed: model.examplesProcessed,
           weightBytes: model.weightBytes, parameters: model.parameters, trainingLoss: model.trainingLoss,
-          evaluation: model.evaluation, controls: model.specification?.commands?.length || 0,
+          evaluation: model.evaluation, validation: model.validation, distinctMinutes: model.distinctMinutes, controls: model.specification?.commands?.length || 0,
           compatible: Boolean(id && compatible(model, roverManager.rovers.get(id))) })),
       catalogError: modelError,
       recording: { ...system.recording, sessions: system.recording.sessions.filter((s) => roverManager.canSeeRover(s.roverId, socket)) },
@@ -50,7 +51,11 @@ function createGateway({ root, driving, getSystemState, activity }) {
         catch (error) { modelError = error.message; }
       }
       if (closed) return;
-      for (const [socket, roverId] of viewers) socket.emit('roverLearning:state', stateFor(socket, roverId));
+      const ordinary = Date.now() - lastOrdinary >= 1000;
+      if (ordinary) lastOrdinary = Date.now();
+      for (const [socket, view] of viewers) {
+        if (ordinary || view.live) socket.volatile.emit('roverLearning:state', stateFor(socket, view.roverId));
+      }
     } finally { refreshing = false; }
   }
 
@@ -59,7 +64,7 @@ function createGateway({ root, driving, getSystemState, activity }) {
       const respond = typeof reply === 'function' ? reply : () => {};
       if (!isVerified(socket)) { respond({ error: 'Verification required' }); return; }
       const roverId = typeof payload?.roverId === 'string' ? payload.roverId : null;
-      viewers.set(socket, roverId);
+      viewers.set(socket, { roverId, live: payload.live === true });
       respond({ state: stateFor(socket, roverId) });
       refresh().catch((error) => activity('Card update failed', { error: error.message }));
     };
@@ -74,7 +79,7 @@ function createGateway({ root, driving, getSystemState, activity }) {
           if (!session || !(driving.canControl(socket, roverId) || session.socketId === socket.id)) throw new Error('You cannot stop this controller');
           await driving.stop(roverId);
         } else throw new Error('Unknown action');
-        viewers.set(socket, roverId);
+        viewers.set(socket, { roverId, live: viewers.get(socket)?.live || false });
         respond({ state: stateFor(socket, roverId) });
       } catch (error) { respond({ error: error.message }); }
       refresh().catch((error) => activity('Card update failed', { error: error.message }));
@@ -94,7 +99,7 @@ function createGateway({ root, driving, getSystemState, activity }) {
 
   io.on('connection', connect);
   for (const socket of io.sockets.sockets.values()) connect(socket);
-  const timer = setInterval(() => { refresh().catch((error) => activity('Card update failed', { error: error.message })); }, 1000);
+  const timer = setInterval(() => { refresh().catch((error) => activity('Card update failed', { error: error.message })); }, 100);
   refresh().catch((error) => activity('Catalog unavailable', { error: error.message }));
   return {
     close() {

server/src/services/roverLearningService/storage.js

@@ -30,6 +30,10 @@ async function recoverRecordings(root) {
       // until ordinary retention removes it, rather than blocking all startup.
       metadata = { version: 1, id: recording.id, startedAt: recording.startedAt };
     }
+    if (metadata.bufferOnly) {
+      await fs.rm(recording.directory, { recursive: true, force: true });
+      continue;
+    }
     if (metadata.endedAt != null) continue;
     await saveMetadata(recording.directory, {
       ...metadata, endedAt: Date.now(), reason: 'interrupted',

server/src/services/roverLearningService/training.js

@@ -6,7 +6,7 @@ const { spawn } = require('node:child_process');
 const { randomInt } = require('node:crypto');
 const { listRecordings } = require('./storage');
 
-function createTrainer({ root, config, logger }) {
+function createTrainer({ root, config, actionThreshold, logger }) {
   const leases = new Set();
   const trainingRoot = path.join(root, 'training');
   let ledger = {};
@@ -57,11 +57,10 @@ function createTrainer({ root, config, logger }) {
     for (const id of Object.keys(ledger)) if (!present.has(id)) delete ledger[id];
     const eligible = [];
     for (const recording of recordings) {
-      if (activeIds.has(recording.id) || ledger[recording.id]?.rejected
-          || (ledger[recording.id]?.rounds || 0) >= config.roundsPerSession) continue;
+      if (activeIds.has(recording.id) || ledger[recording.id]?.rejected) continue;
       try {
         const metadata = JSON.parse(await fs.readFile(path.join(recording.directory, 'session.json'), 'utf8'));
-        if (!metadata.endedAt || metadata.error || metadata.droppedEvents || metadata.reason === 'interrupted') continue;
+        if (metadata.bufferOnly || !metadata.endedAt || metadata.error || metadata.droppedEvents || metadata.reason === 'interrupted') continue;
         eligible.push(recording.id);
       } catch (error) {
         ledger[recording.id] = { rejected: `Unreadable session metadata: ${error.message}` };
@@ -76,15 +75,15 @@ function createTrainer({ root, config, logger }) {
     const selected = eligible.slice(0, 4);
     nextRun = Date.now() + config.intervalSeconds * 1000;
     if (!selected.length) {
-      state = { ...state, status: 'waiting' };
+      state = { ...state, status: 'waiting', reason: 'Waiting for completed usable human recordings' };
       await saveLedger();
       return;
     }
     const jobPath = path.join(trainingRoot, 'job.json');
-    await fs.writeFile(jobPath, JSON.stringify({ root, config, sessions: selected }));
+    await fs.writeFile(jobPath, JSON.stringify({ root, config, actionThreshold, sessions: selected }));
     if (stopped) return;
     selected.forEach((id) => leases.add(id));
-    state = { ...state, status: 'training', sessions: selected };
+    state = { ...state, status: 'training', reason: null, sessions: selected };
     state.startedAt = Date.now();
     child = spawn(config.python, [path.join(__dirname, 'workers/train.py'), '--job', jobPath], {
       stdio: ['ignore', 'pipe', 'pipe'],
@@ -138,7 +137,9 @@ function createTrainer({ root, config, logger }) {
             }
             state = { ...state, status: stopped ? 'stopped' : 'waiting', steps: result.steps ?? state.steps,
               completedAt: Date.now(), elapsedSeconds: result.elapsedSeconds ?? null, error: null,
-              examples: result.examples ?? state.examples, loss: result.loss ?? null, reason: result.reason ?? null };
+              examples: result.examples ?? state.examples, loss: result.loss ?? null,
+              publication: result.publication ?? state.publication, evaluation: result.evaluation ?? state.evaluation,
+              reason: result.reason ?? 'Waiting for the next scheduled training job' };
             await saveLedger();
             logger.info('Training job completed', { ...state, model: result.published?.name || null });
           }

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

@@ -109,65 +109,74 @@ def windows(directory, schema, cancelled=lambda: False):
     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
+    def frames():
+        files = metadata['video'].get('files', ['video.mkv'])
+        for name in files:
+            if Path(name).name != name:
+                raise ValueError('Invalid video segment path')
+            with av.open(str(directory / name)) as container:
+                stream = container.streams.video[0]
+                stream.codec_context.thread_count = 1
+                yield from container.decode(stream)
+
+    for frame in frames():
+        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 < metadata.get('trainingStartedAt', metadata['startedAt']):
+            continue
+        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

@@ -110,10 +110,8 @@ def main():
                 raise ValueError('Nonfinite model output')
             families = {}
             active = set()
+            proposals = []
             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':
@@ -127,6 +125,11 @@ def main():
                     for key in field['path'][:-1]:
                         target = target[key]
                     target[field['path'][-1]] = round(number) if field['integer'] else number
+                proposals.append({'slot': slot, 'score': probabilities[slot], 'command': command,
+                                  'reason': 'Below threshold'})
+                if probabilities[slot] < args.threshold:
+                    continue
+                active.add(slot)
                 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)
@@ -135,12 +138,18 @@ def main():
             # 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]
+            winners = {row[2]: row for row in families.values()}
+            for proposal in proposals:
+                slot = proposal['slot']
+                if slot in active:
+                    proposal['reason'] = ('Competing action' if slot not in winners else
+                                          'Held discrete action' if winners[slot][3] and slot in latched else 'Proposed')
             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)
+                              'commands': commands, 'proposals': proposals, 'threshold': args.threshold}, allow_nan=False), flush=True)
 
 
 if __name__ == '__main__':

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

@@ -69,10 +69,13 @@ def main():
     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'
+    # A fresh training lineage prevents previously trained footage leaking into validation.
+    resume_path = root / 'training' / 'resume-v2.pt'
     schema, steps, examples, last_published = [], 0, 0, 0
     trained_rovers = set()
     control_profiles = []
+    distinct_windows = {}
+    published_windows = 0
     if resume_path.exists():
         saved = torch.load(resume_path, map_location='cpu', weights_only=True)
         if saved['version'] != VERSION:
@@ -84,10 +87,15 @@ def main():
         last_published = saved['lastPublished']
         trained_rovers = set(saved['trainedRovers'])
         control_profiles = saved.get('controlProfiles', [])
+        distinct_windows = saved.get('distinctWindows', {})
+        published_windows = saved.get('publishedWindows', 0)
     initial_steps = steps
-    pool, accepted, rejected = [], [], {}
+    pool, validation, accepted, rejected = [], [], [], {}
+    observed_windows = {}
     sessions = list(job['sessions'])
     random.shuffle(sessions)
+    is_held_out = lambda session_id: int(hashlib.sha256(session_id.encode()).hexdigest()[:8], 16) % 5 == 0
+    sessions.sort(key=is_held_out)
     per_session = max(1, config['maxSamples'] // max(1, len(sessions)))
     for session_id in sessions:
         if cancelled():
@@ -110,13 +118,23 @@ def main():
                 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)
+            # Entire sessions stay in one partition across all jobs. Neighboring
+            # video windows must never straddle training and validation.
+            held_out = is_held_out(session_id)
+            if held_out:
+                if candidate_schema != schema:
+                    continue  # Retry after training data introduces these command shapes.
+                validation.extend(samples)
+            else:
+                schema = candidate_schema
+                pool.extend(samples)
+                observed_windows[session_id] = seen
             accepted.append(session_id)
             metadata = json.loads((directory / 'session.json').read_text())
-            trained_rovers.add(str(metadata['roverId']))
+            if not held_out:
+                trained_rovers.add(str(metadata['roverId']))
             profile = metadata.get('controlProfile')
-            if profile is not None and profile not in control_profiles:
+            if not held_out and 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:
@@ -124,7 +142,7 @@ def main():
             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')
+             reason='Stopped/time limit' if cancelled() else 'Waiting for enough training-partition samples with human actions')
         return
 
     numeric_mask = torch.zeros(SLOTS, VALUES)
@@ -169,8 +187,57 @@ def main():
         emit('complete', trained=False, accepted=[], rejected=rejected, reason='No training steps completed')
         return
 
+    distinct_windows.update(observed_windows)
+    unique_count = sum(distinct_windows.values())
+    evaluation = {'status': 'waiting for held-out sessions', 'windows': len(validation)}
+    if validation and not cancelled():
+        model.eval()
+        true_positive = predicted_positive = actual_positive = 0
+        value_error = value_count = 0
+        evaluated = 0
+        with torch.inference_mode():
+            for offset in range(0, len(validation), config['batchSize']):
+                if cancelled():
+                    break
+                batch = validation[offset:offset + config['batchSize']]
+                images, states, targets, values, available = [torch.from_numpy(np.stack(items)) for items in zip(*batch)]
+                logits, predictions = model(images.float() / 255, states)
+                predicted = (logits.sigmoid() >= job['actionThreshold']) & available.bool()
+                actual = targets.bool() & available.bool()
+                true_positive += (predicted & actual).sum().item()
+                predicted_positive += predicted.sum().item()
+                actual_positive += actual.sum().item()
+                mask = numeric_mask.unsqueeze(0) * targets.unsqueeze(-1)
+                value_error += ((predictions - values).square() * mask).sum().item()
+                value_count += mask.sum().item()
+                evaluated += len(batch)
+        precision = true_positive / max(1, predicted_positive)
+        recall = true_positive / max(1, actual_positive)
+        mse = value_error / max(1, value_count)
+        # A basic imitation gate, not an autonomous-driving certification.
+        passed = (evaluated == len(validation) and evaluated >= 128 and actual_positive >= 32
+                  and precision >= .5 and recall >= .5 and value_count > 0 and mse <= .1)
+        evaluation = {'status': 'passed imitation gate' if passed else 'imitation gate not met',
+                      'windows': evaluated, 'precision': precision, 'recall': recall,
+                      'numericMse': mse, 'positiveActions': actual_positive}
+    publication = {'distinctMinutes': unique_count / 600,
+                   'newMinutes': (unique_count - published_windows) / 600,
+                   'minimumMinutes': config['minimumDrivingMinutes'],
+                   'requiredNewMinutes': config['newDrivingMinutesPerModel']}
+    reasons = []
+    if publication['distinctMinutes'] < config['minimumDrivingMinutes']:
+        reasons.append('Collecting distinct driving footage')
+    if publication['newMinutes'] < config['newDrivingMinutesPerModel']:
+        reasons.append('Waiting for new driving footage since last publication')
+    if steps - last_published < config['checkpointEverySteps']:
+        reasons.append('Training steps below publication interval')
+    if evaluation['status'] != 'passed imitation gate':
+        reasons.append(evaluation['status'])
+    if cancelled():
+        reasons.append('Job time budget reached')
+    publication['reason'] = '; '.join(reasons) or 'Published'
     published = None
-    if steps - last_published >= config['checkpointEverySteps']:
+    if not reasons:
         adjectives = ['amber', 'curious', 'gentle', 'quiet', 'silver', 'bright', 'merry', 'sleepy']
         animals = ['otter', 'finch', 'badger', 'robin', 'fox', 'heron', 'marten', 'wren']
         model_id = str(uuid.uuid4())
@@ -188,22 +255,24 @@ def main():
             '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__),
+            'evaluation': evaluation['status'], 'validation': evaluation, 'distinctMinutes': publication['distinctMinutes'], '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
+        published_windows = unique_count
     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,
+        'controlProfiles': control_profiles, 'distinctWindows': distinct_windows,
+        'publishedWindows': published_windows,
     }, 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)
+         published=published, publication=publication, evaluation=evaluation, elapsedSeconds=time.monotonic() - started)
 
 
 if __name__ == '__main__':