slop

server/src/services/roverLearningService/capabilities.js

@@ -7,7 +7,7 @@ function controlProfile(meta = {}) {
 }
 
 function compatible(model, record) {
-  if (model?.specification?.version !== 1 || !record) return false;
+  if (model?.specification?.version !== 2 || !record) return false;
   const profile = controlProfile(record.meta);
   return Array.isArray(model.controlProfiles) && model.controlProfiles.some((saved) => isDeepStrictEqual(saved, profile));
 }

server/src/services/roverLearningService/diagnostics.js

@@ -0,0 +1,60 @@
+// Export an explicit allowlist, never the raw socket/session/configuration object.
+const os = require('node:os');
+const fs = require('node:fs');
+const path = require('node:path');
+const { createHash } = require('node:crypto');
+const applicationVersion = require('../../../package.json').version;
+const hash = createHash('sha256');
+for (const file of ['workers/policy.py', 'workers/dataset.py', 'workers/train.py', 'workers/infer.py',
+  'driving.js', 'recording.js', 'training.js', 'socketGateway.js', 'diagnostics.js']) {
+  hash.update(fs.readFileSync(path.join(__dirname, file)));
+}
+const implementationSha256 = hash.digest('hex');
+
+function errorCategory(value) {
+  if (!value) return null;
+  const text = String(value).toLowerCase();
+  for (const [pattern, category] of [
+    [/timestamp|pts|clock|aligned/, 'time_alignment'], [/sensor/, 'sensors'],
+    [/video|decode|ffmpeg|rtsp/, 'video'], [/permission|turn|control permission/, 'permission'],
+    [/schema|shape|version/, 'model_schema'], [/space|disk|enospc/, 'storage'],
+    [/sample|window|command/, 'dataset_or_command'], [/worker|exited|process/, 'worker'],
+  ]) if (pattern.test(text)) return category;
+  return 'unclassified';
+}
+
+function report(state, controller) {
+  const training = state.training || {};
+  const metrics = training.diagnostics || {};
+  const session = state.session;
+  return {
+    reportVersion: 1, generatedAtUnixMs: state.updatedAt,
+    software: { applicationVersion, implementationSha256, policyVersion: 2, node: process.version,
+      dependencies: metrics.dependencies || null },
+    host: { logicalCpus: os.cpus().length, totalMemoryBytes: os.totalmem(), freeMemoryBytes: os.freemem(), loadAverage: os.loadavg() },
+    settings: state.limits,
+    recording: { paused: state.recording.paused, bytes: state.recording.bytes,
+      freeBytes: state.recording.freeBytes, pendingBytes: state.recording.pendingBytes,
+      droppedPending: state.recording.droppedPending,
+      sessions: state.recording.sessions.map((item) => ({ startedAtUnixMs: item.startedAt,
+        status: item.video, droppedEvents: item.droppedEvents, errorCategory: errorCategory(item.error) })) },
+    training: { status: training.status, steps: training.steps, examples: training.examples,
+      loss: training.loss, losses: training.losses || null, publication: training.publication || null,
+      evaluation: training.evaluation || null, dataset: metrics.dataset || training.dataset || null,
+      resources: { decodeSeconds: metrics.decodeSeconds ?? null, trainingSeconds: metrics.trainingSeconds ?? null,
+        evaluationSeconds: metrics.evaluationSeconds ?? null, checkpointSeconds: metrics.checkpointSeconds ?? null, cpuSeconds: metrics.cpuSeconds ?? null,
+        peakRssMiB: metrics.peakRssMiB ?? null },
+      history: training.history || [], rejectionCategories: (training.rejectionReasons || []).map(errorCategory),
+      errorCategory: errorCategory(training.error), nextAttemptAtUnixMs: training.nextAttemptAt },
+    model: session ? { id: session.modelId, threshold: session.threshold, appliedThreshold: session.appliedThreshold } : null,
+    controller: controller ? { modelId: controller.modelId, sensorPresent: controller.sensorPresent || null,
+      stopCategory: errorCategory(controller.stopReason), history: (controller.history || []).map((item) => ({
+        ...item, submissions: item.submissions.map((submission) => ({ type: submission.type,
+          wheels: submission.wheels, result: submission.result === 'issued' ? 'issued' : errorCategory(submission.result) })) })) } : null,
+    sensorFields: metrics.policy?.sensorFields || null,
+    timing: { alignment: 'server_receipt', browserVideoLatencyMs: null, wheelCommandExpiryMs: null },
+    events: state.activity.map(({ at, message }) => ({ atUnixMs: at, message })),
+    unavailable: ['browser_video_latency', 'actual_physical_command_execution', 'live_training_cpu_utilization'],
+  };
+}
+module.exports = { report };

server/src/services/roverLearningService/driving.js

@@ -19,6 +19,7 @@ function canControl(socket, roverId) {
 
 function createDriving({ root, config, onChange, activity }) {
   const sessions = new Map();
+  const recent = new Map();
   let stopping = false;
   let watchdog;
   let unsubscribe;
@@ -35,7 +36,8 @@ function createDriving({ root, config, onChange, activity }) {
     if (!OPERATING.has(command?.type)) throw new Error('Unsupported model command');
     const record = roverManager.rovers.get(session.roverId);
     if (command.type === 'drive') {
-      command.driveDirect = { left: clamp(command.driveDirect?.left, -500, 500), right: clamp(command.driveDirect?.right, -500, 500) };
+      const limit = Math.min(500, Number(record.meta?.maxWheelSpeed) || 500);
+      command.driveDirect = { left: Math.round(clamp(command.driveDirect?.left, -limit, limit)), right: Math.round(clamp(command.driveDirect?.right, -limit, limit)) };
     }
     if (command.type === 'motors') {
       command.motorPwm = { main: clamp(command.motorPwm?.main, -127, 127), side: clamp(command.motorPwm?.side, -127, 127), vacuum: clamp(command.motorPwm?.vacuum, 0, 127) };
@@ -69,6 +71,9 @@ function createDriving({ root, config, onChange, activity }) {
     const timer = setTimeout(() => session.worker.kill('SIGKILL'), 2000);
     session.done = session.closed.then(() => {
       clearTimeout(timer);
+      recent.delete(roverId);
+      recent.set(roverId, { modelId: session.modelId, stopReason: reason, history: session.history });
+      if (recent.size > 20) recent.delete(recent.keys().next().value);
       sessions.delete(roverId);
       activity('Controller stopped', { roverId, reason });
       onChange();
@@ -90,9 +95,10 @@ function createDriving({ root, config, onChange, activity }) {
     const worker = spawn(config.training.python, [path.join(__dirname, 'workers/infer.py'), '--model', directory,
       '--url', `rtsp://127.0.0.1:8554/${encodeURIComponent(roverId)}`, '--threads', String(config.driving.threads),
       '--threshold', String(config.driving.actionThreshold)], { stdio: ['pipe', 'pipe', 'pipe'], env: { ...process.env, PYTHONDONTWRITEBYTECODE: '1' } });
-    const session = { worker, socket, roverId, modelId, modelName: metadata.name, status: 'starting', startedAt: Date.now(), lastPredictionAt: null };
+    const session = { worker, socket, roverId, modelId, modelName: metadata.name, status: 'starting', startedAt: Date.now(), lastPredictionAt: null, threshold: config.driving.actionThreshold, history: [] };
     session.closed = new Promise((resolve) => worker.once('close', resolve));
     sessions.set(roverId, session);
+    recent.delete(roverId);
     let output = '';
     let stderr = '';
     worker.stdin.on('error', () => { stop(roverId, 'Worker input closed'); });
@@ -120,12 +126,24 @@ function createDriving({ root, config, onChange, activity }) {
           session.sensorAt = message.sensorAt;
           session.latencyMs = message.latencyMs;
           session.proposals = Array.isArray(message.proposals) ? message.proposals.slice(0, 64) : [];
-          session.threshold = message.threshold;
+          session.appliedThreshold = message.threshold;
+          session.wheelSpeeds = message.wheelSpeeds;
+          session.sensorPresent = message.sensorPresent;
           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';
+          if (!Array.isArray(message.commands) || message.commands.length > 65) throw new Error('Invalid action batch');
+          session.predictionStatus = message.wheelSpeeds?.some((value) => value !== 0) ? 'Sending wheel controls' : 'Predicting stopped wheels';
           // Manual commands intentionally do not stop or pause this controller.
           for (const command of message.commands) transmit(session, command);
+          // Bound report history independently of camera and UI frame rates.
+          if (!session.history.length || Date.now() - session.history.at(-1).at >= 500) {
+            session.history.push({ at: Date.now(), wheelSpeeds: session.wheelSpeeds,
+              latencyMs: session.latencyMs, frameAgeMs: Date.now() - session.frameAt,
+              sensorAgeMs: Date.now() - session.sensorAt, threshold: session.appliedThreshold,
+              proposals: [...session.proposals].sort((a, b) => b.score - a.score).slice(0, 8).map(({ slot, score, reason }) => ({ slot, score, reason })),
+              submissions: session.submissions.slice(0, 16).map(({ command, result }) => ({ type: command.type, result,
+                wheels: command.type === 'drive' ? command.driveDirect : undefined })) });
+            if (session.history.length > 60) session.history.shift();
+          }
         } catch (error) { stop(roverId, error.message); break; }
       }
     });
@@ -158,11 +176,22 @@ function createDriving({ root, config, onChange, activity }) {
 
   return {
     attach, start, stop, canControl,
+    setThreshold(socket, roverId, value) {
+      const session = sessions.get(roverId);
+      if (!session || session.status === 'stopping' || !canControl(socket, roverId)) throw new Error('You cannot adjust this controller');
+      if (typeof value !== 'number' || !Number.isFinite(value) || value < .05 || value > .99) throw new Error('Threshold must be between 0.05 and 0.99');
+      session.threshold = value;
+      sendState(session, { kind: 'settings', ts: Date.now(), threshold: value });
+    },
+    getDiagnostics(id) {
+      const session = sessions.get(id);
+      return session ? { modelId: session.modelId, sensorPresent: session.sensorPresent, history: session.history } : recent.get(id) || null;
+    },
     isActive: (id) => sessions.has(id),
     getState: (id) => {
       const session = sessions.get(id);
       if (!session) return null;
-      const { worker, socket, closed, done, ...state } = session;
+      const { worker, socket, closed, done, history, ...state } = session;
       return { ...state, startedBy: socket.data?.nickname || socket.id, socketId: socket.id };
     },
     shutdown: async () => {

server/src/services/roverLearningService/recording.js

@@ -21,7 +21,7 @@ async function createRecording({ root, roverId, snapshot, logger, docked = false
   await fsp.mkdir(directory);
   const metadata = {
     version: 1, id, roverId, startedAt: Date.now(), endedAt: null,
-    controlProfile: controlProfile(snapshot.meta),
+    controlProfile: controlProfile(snapshot.meta), wheelSpeedLimit: snapshot.meta?.maxWheelSpeed ?? 500,
     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, bufferOnly: docked,

server/src/services/roverLearningService/socketGateway.js

@@ -3,6 +3,7 @@ const io = require('../../globals/io');
 const roverManager = require('../roverManager');
 const { isVerified } = require('../verificationService');
 const { listModels } = require('./models');
+const { report } = require('./diagnostics');
 const { compatible } = require('./capabilities');
 
 function createGateway({ root, driving, getSystemState, activity }) {
@@ -22,16 +23,16 @@ function createGateway({ root, driving, getSystemState, activity }) {
     const session = id ? driving.getState(id) : null;
     const canControl = Boolean(id && driving.canControl(socket, id));
     const system = getSystemState();
-    return {
+    const state = {
       available: true, roverId: id, updatedAt: Date.now(),
-      control: { canStart: canControl && !session, canStop: Boolean(session && (canControl || session.socketId === socket.id)),
+      control: { canStart: canControl && !session, canAdjust: canControl && Boolean(session), canStop: Boolean(session && (canControl || session.socketId === socket.id)),
         reason: !id ? 'Select a rover first' : !canControl ? 'You do not currently have control of this rover' : null },
       session,
       models: models.filter((model) => model.trainedRovers?.some((rover) => roverManager.canSeeRover(rover, socket)))
         .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, validation: model.validation, distinctMinutes: model.distinctMinutes, controls: model.specification?.commands?.length || 0,
+          formatVersion: model.specification?.version, 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)) },
@@ -39,6 +40,8 @@ function createGateway({ root, driving, getSystemState, activity }) {
       limits: system.limits,
       activity: system.activity.filter((entry) => !entry.roverId || roverManager.canSeeRover(entry.roverId, socket)),
     };
+    state.diagnostics = viewers.get(socket)?.live ? report(state, id ? driving.getDiagnostics(id) : null) : null;
+    return state;
   }
 
   async function refresh() {
@@ -54,7 +57,7 @@ function createGateway({ root, driving, getSystemState, activity }) {
       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));
+        if (ordinary || view.live || driving.isActive(view.roverId)) socket.volatile.emit('roverLearning:state', stateFor(socket, view.roverId));
       }
     } finally { refreshing = false; }
   }
@@ -74,6 +77,7 @@ function createGateway({ root, driving, getSystemState, activity }) {
         if (!isVerified(socket)) throw new Error('Verification required');
         const roverId = String(payload.roverId || '');
         if (payload.action === 'start') await driving.start(socket, roverId, String(payload.modelId || ''));
+        else if (payload.action === 'threshold') driving.setThreshold(socket, roverId, payload.value);
         else if (payload.action === 'stop') {
           const session = driving.getState(roverId);
           if (!session || !(driving.canControl(socket, roverId) || session.socketId === socket.id)) throw new Error('You cannot stop this controller');

server/src/services/roverLearningService/training.js

@@ -10,6 +10,7 @@ function createTrainer({ root, config, actionThreshold, logger }) {
   const leases = new Set();
   const trainingRoot = path.join(root, 'training');
   let ledger = {};
+  const history = [];
   let child = null;
   let completion = Promise.resolve();
   let stopped = false;
@@ -30,7 +31,7 @@ function createTrainer({ root, config, actionThreshold, logger }) {
         }
       }
       try {
-        const saved = JSON.parse(await fs.readFile(path.join(trainingRoot, 'schedule.json'), 'utf8'));
+        const saved = JSON.parse(await fs.readFile(path.join(trainingRoot, 'schedule-v3.json'), 'utf8'));
         ledger = saved.sessions;
         state = { ...saved.progress, status: 'waiting' };
         if (!ledger || typeof ledger !== 'object') throw new Error('Invalid training schedule');
@@ -43,7 +44,7 @@ function createTrainer({ root, config, actionThreshold, logger }) {
   }
 
   async function saveLedger() {
-    const filename = path.join(trainingRoot, 'schedule.json');
+    const filename = path.join(trainingRoot, 'schedule-v3.json');
     await fs.writeFile(`${filename}.tmp`, JSON.stringify({ sessions: ledger, progress: state }));
     await fs.rename(`${filename}.tmp`, filename);
   }
@@ -117,8 +118,8 @@ function createTrainer({ root, config, actionThreshold, logger }) {
           if (message.kind === 'complete') result = message;
           if (message.kind === 'error') errorMessage = message.message;
           if (message.kind === 'skipped') logger.warn('Training skipped recording', message);
-          if (message.kind === 'dataset') state = { ...state, lastSampledWindows: message.sampled, lastAvailableWindows: message.windows };
-          if (message.kind === 'progress') state = { ...state, steps: message.steps, examples: message.examples, loss: message.loss };
+          if (message.kind === 'dataset') state = { ...state, lastSampledWindows: message.sampled, lastAvailableWindows: message.windows, dataset: message.diagnostics };
+          if (message.kind === 'progress') state = { ...state, steps: message.steps, examples: message.examples, loss: message.loss, losses: message.losses };
         } catch { errorMessage = 'Invalid training worker message'; }
       }
     });
@@ -138,7 +139,7 @@ function createTrainer({ root, config, actionThreshold, 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,
-              publication: result.publication ?? state.publication, evaluation: result.evaluation ?? state.evaluation,
+              diagnostics: result.diagnostics ?? state.diagnostics, 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 });
@@ -147,6 +148,10 @@ function createTrainer({ root, config, actionThreshold, logger }) {
           state = { ...state, status: 'error', error: error.message };
           logger.error('Training result persistence failed', { error: error.message });
         } finally {
+          history.push({ at: Date.now(), status: state.status, steps: state.steps, loss: state.loss,
+            losses: state.losses || null, evaluation: state.evaluation || null,
+            elapsedSeconds: state.elapsedSeconds || null });
+          if (history.length > 20) history.shift();
           leases.clear();
           child = null;
           nextRun = Date.now() + config.intervalSeconds * 1000;
@@ -166,7 +171,7 @@ function createTrainer({ root, config, actionThreshold, logger }) {
 
   return {
     leases, start, tick, cancel,
-    getState: () => ({ ...state, nextAttemptAt: nextRun || null, leasedSessions: leases.size,
+    getState: () => ({ ...state, history, nextAttemptAt: nextRun || null, leasedSessions: leases.size,
       visitedSessions: Object.values(ledger).filter((entry) => entry.rounds).length,
       skippedSessions: Object.values(ledger).filter((entry) => entry.rejected).length,
       rejectionReasons: [...new Set(Object.values(ledger).filter((entry) => entry.rejected).map((entry) => entry.rejected))].slice(0, 10) }),

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

@@ -47,7 +47,7 @@ def load_events(directory):
     return metadata, sorted(events, key=lambda e: e['ts'])
 
 
-def prepare(directory, schema):
+def prepare(directory, schema, stats):
     metadata, events = load_events(directory)
     requests = {}
     accepted = set()
@@ -72,9 +72,28 @@ def prepare(directory, schema):
     if not sensors:
         raise ValueError('No sensor observations')
     candidate = copy.deepcopy(schema)
-    history, targets = [], []
+    history, targets, wheels = [], [], [(stamp, None) for stamp in boundaries]
+    stats.update({'sensorFrames': len(sensors), 'dispatches': len(dispatches),
+                  'failedCommands': len(failed), 'sensorPresentFraction': np.mean([row[1][32:] for row in sensors], axis=0).tolist()})
     for data in dispatches:
-        encoded = encode_command(data.get('command') or {}, candidate, extend=True)
+        command = data.get('command') or {}
+        request = requests.get(data.get('requestId'))
+        human = (data.get('source') == 'client' and request and data.get('commandId') in accepted
+                 and data.get('commandId') not in failed)
+        stamp = data['ts']
+        if command.get('type') == 'drive':
+            direct = command.get('driveDirect') or {}
+            pair = [direct.get('left'), direct.get('right')]
+            if human and all(isinstance(v, (int, float)) and math.isfinite(v) for v in pair):
+                wheels.append((stamp, np.clip(np.array(pair, dtype=np.float32), -metadata.get('wheelSpeedLimit', 500), metadata.get('wheelSpeedLimit', 500)) / 500))
+            else:
+                wheels.append((stamp, None))
+        elif command.get('type') in ('raw', 'reset', 'reboot', 'update', 'sensorStream'):
+            # Raw OI can change mode or motion; do not assume the last wheel
+            # command survives it. A later explicit human drive resolves intent.
+            wheels.append((stamp, None))
+            boundaries.append(stamp)
+        encoded = encode_command(command, candidate, extend=True)
         if encoded is None:
             continue
         stamp = data['ts']
@@ -93,17 +112,22 @@ def prepare(directory, schema):
     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)
+    return metadata, sensors, sorted(wheels, key=lambda x: x[0]), sorted(history, key=lambda x: x[0]), sorted(targets, key=lambda x: x[0]), sorted(boundaries)
 
 
-def windows(directory, schema, cancelled=lambda: False):
+def windows(directory, schema, cancelled=lambda: False, stats=None):
     directory = Path(directory)
-    metadata, sensors, history, targets, boundaries = prepare(directory, schema)
+    stats = stats if stats is not None else {}
+    metadata, sensors, wheels, history, targets, boundaries = prepare(directory, schema, stats)
+    wheel_times = [row[0] for row in wheels]
+    stats.update({'usableWindows': 0, 'movingWindows': 0, 'stoppedWindows': 0,
+                  'unknownWheelWindows': 0, 'staleSensorWindows': 0, 'boundaryWindows': 0,
+                  'ambiguousWindows': 0, 'videoFrames': 0, 'maxVideoGapMs': 0})
     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
+        available[slot] = float(schema[slot]['template']['type'] != 'drive')
     command_values = np.zeros((SLOTS, VALUES), dtype=np.float32)
     command_times = np.full(SLOTS, -np.inf)
     cursor = 0
@@ -127,6 +151,9 @@ def windows(directory, schema, cancelled=lambda: False):
         stamp = float(frame.pts * frame.time_base) * 1000
         if not math.isfinite(stamp) or stamp < last_pts:
             raise ValueError('Invalid/nonmonotonic video timestamps')
+        stats['videoFrames'] += 1
+        if math.isfinite(last_pts):
+            stats['maxVideoGapMs'] = max(stats['maxVideoGapMs'], stamp - last_pts)
         last_pts = stamp
         if not metadata['startedAt'] - 2000 <= stamp <= metadata['endedAt'] + 2000:
             raise ValueError('Video PTS is not aligned to session Unix time')
@@ -142,6 +169,7 @@ def windows(directory, schema, cancelled=lambda: False):
             cursor += 1
         sensor_index = bisect.bisect_right(sensor_times, stamp) - 1
         if sensor_index < 0 or stamp - sensor_times[sensor_index] > 1000:
+            stats['staleSensorWindows'] += 1
             recent.clear()
             continue
         seen = np.isfinite(command_times).astype(np.float32)
@@ -158,7 +186,14 @@ def windows(directory, schema, cancelled=lambda: False):
         if len(chosen) != 4 or stamp + 100 > metadata['endedAt']:
             continue
         if bisect.bisect_right(boundaries, stamp + 100) > bisect.bisect_left(boundaries, chosen[0][0]):
+            stats['boundaryWindows'] += 1
             continue
+        before = bisect.bisect_left(wheel_times, stamp) - 1
+        after = bisect.bisect_right(wheel_times, stamp + 100) - 1
+        if before < 0 or after < 0 or wheels[before][1] is None or wheels[after][1] is None:
+            stats['unknownWheelWindows'] += 1
+            continue
+        wheel_target, previous_wheels = wheels[after][1], wheels[before][1]
         start = bisect.bisect_left(target_times, stamp)
         end = bisect.bisect_left(target_times, stamp + 100)
         actions = np.zeros(SLOTS, dtype=np.float32)
@@ -168,6 +203,8 @@ def windows(directory, schema, cancelled=lambda: False):
         for _, slot, vector in targets[start:end]:
             template = schema[slot]['template']
             family = template['type']
+            if family == 'drive':
+                continue
             if family == 'peripheral':
                 peripheral = template.get('peripheral') or {}
                 family += ':' + str(peripheral.get('id')) + ':' + str(peripheral.get('control'))
@@ -178,5 +215,8 @@ def windows(directory, schema, cancelled=lambda: False):
             families.add(family)
             actions[slot], values[slot] = 1, vector
         if ambiguous:
+            stats['ambiguousWindows'] += 1
             continue
-        yield np.stack([row[1] for row in chosen]), np.stack([row[2] for row in chosen]), actions, values, available
+        stats['usableWindows'] += 1
+        stats['movingWindows' if np.max(np.abs(wheel_target)) > .02 else 'stoppedWindows'] += 1
+        yield np.stack([row[1] for row in chosen]), np.stack([row[2] for row in chosen]), actions, values, available, wheel_target, previous_wheels

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

@@ -74,7 +74,9 @@ def main():
             with mutex:
                 while observations and observations[0]['ts'] < stamp:
                     event = observations.popleft()
-                    if event['kind'] == 'sensor':
+                    if event['kind'] == 'settings':
+                        args.threshold = max(.05, min(.99, float(event['threshold'])))
+                    elif event['kind'] == 'sensor':
                         sensors, sensor_time = event['data'], event['ts']
                     elif event['kind'] == 'command':
                         try:
@@ -102,16 +104,19 @@ def main():
                 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,
+                logits, numeric, wheel_output = 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()
+                wheel_speeds = [round(value * 500) for value in wheel_output[0].tolist()]
             if not all(math.isfinite(value) for value in probabilities):
                 raise ValueError('Nonfinite model output')
             families = {}
             active = set()
             proposals = []
             for slot, descriptor in enumerate(schema):
+                if descriptor['template']['type'] == 'drive':
+                    continue
                 command = json.loads(json.dumps(descriptor['template']))
                 family = command['type']
                 if family == 'peripheral':
@@ -147,8 +152,10 @@ def main():
             for _, _, slot, discrete in families.values():
                 if discrete:
                     latched.add(slot)
+            commands.insert(0, {'type': 'drive', 'driveDirect': dict(zip(('left', 'right'), wheel_speeds))})
             print(json.dumps({'kind': 'prediction', 'frameAt': stamp, 'sensorAt': sensor_time,
                               'latencyMs': (time.perf_counter() - before) * 1000,
+                              'wheelSpeeds': wheel_speeds, 'sensorPresent': sensor_vector(sensors)[32:].tolist(),
                               'commands': commands, 'proposals': proposals, 'threshold': args.threshold}, allow_nan=False), flush=True)
 
 

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

@@ -12,7 +12,7 @@ import numpy as np
 import torch
 from torch import nn
 
-VERSION = 1
+VERSION = 2
 SLOTS = 64
 VALUES = 8
 FRAMES = 4
@@ -109,6 +109,7 @@ class Policy(nn.Module):
             nn.Linear(128 * 3 * 4, 128), nn.ReLU(),
         )
         self.history = nn.GRU(128 + STATE_SIZE, 128, batch_first=True)
+        self.wheels = nn.Linear(128, 2)
         self.actions = nn.Linear(128, SLOTS)
         self.values = nn.Linear(128, SLOTS * VALUES)
 
@@ -117,15 +118,17 @@ class Policy(nn.Module):
         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()
+        return self.actions(last), self.values(last).reshape(batch, SLOTS, VALUES).tanh(), self.wheels(last).tanh()
 
 
 def specification(schema):
-    return {'version': VERSION, 'architecture': 'spatial-cnn-gru-v1',
+    return {'version': VERSION, 'architecture': 'spatial-cnn-gru-wheels-v2',
             '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',
+            'commands': schema, 'actionSemantics': 'Continuous wheels at +100ms; accessories are events in the next 100ms',
+            'wheelScaleMmPerSecond': 500, 'wheelOrder': ['left', 'right'],
+            'wheelCommandExpiryMs': None, 'unknownWheelIntent': 'Excluded until next accepted human drive',
             'timing': 'Server-receipt alignment; browser display latency unmeasured'}

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

@@ -11,6 +11,7 @@ import json
 import os
 from pathlib import Path
 import random
+import resource
 import signal
 import time
 import uuid
@@ -70,7 +71,7 @@ def main():
     model = Policy().cpu()
     optimizer = torch.optim.AdamW(model.parameters(), lr=0.0001)
     # A fresh training lineage prevents previously trained footage leaking into validation.
-    resume_path = root / 'training' / 'resume-v2.pt'
+    resume_path = root / 'training' / 'resume-v3.pt'
     schema, steps, examples, last_published = [], 0, 0, 0
     trained_rovers = set()
     control_profiles = []
@@ -89,6 +90,9 @@ def main():
         control_profiles = saved.get('controlProfiles', [])
         distinct_windows = saved.get('distinctWindows', {})
         published_windows = saved.get('publishedWindows', 0)
+    cpu_started = time.process_time()
+    decode_started = time.monotonic()
+    dataset_stats = []
     initial_steps = steps
     pool, validation, accepted, rejected = [], [], [], {}
     observed_windows = {}
@@ -102,18 +106,34 @@ def main():
             break
         candidate_schema = copy.deepcopy(schema)
         samples, seen = [], 0
+        stats = {}
+        buckets = {key: [] for key in ('stopped', 'forward', 'reverse', 'turn')}
+        counts = dict.fromkeys(buckets, 0)
+        held_out = is_held_out(session_id)
         try:
             directory = root / 'recordings' / session_id
-            for sample in windows(directory, candidate_schema, cancelled):
+            for sample in windows(directory, candidate_schema, cancelled, stats):
                 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)
+                # Keep evaluation naturally distributed. Training reserves space
+                # for ordinary turns/reversing so idle or straight motion cannot
+                # consume the entire sample budget. No scripted driving required.
+                if held_out:
+                    reservoir, count, capacity = samples, seen, per_session
                 else:
-                    index = random.randrange(seen)
-                    if index < per_session:
-                        samples[index] = sample
+                    left, right = sample[5]
+                    category = ('stopped' if max(abs(left), abs(right)) <= .02 else
+                                'turn' if abs(left - right) > .1 else
+                                'reverse' if left + right < 0 else 'forward')
+                    counts[category] += 1
+                    reservoir, count, capacity = buckets[category], counts[category], max(1, per_session // 4)
+                if len(reservoir) < capacity:
+                    reservoir.append(sample)
+                else:
+                    index = random.randrange(count)
+                    if index < capacity:
+                        reservoir[index] = sample
+            if not held_out:
+                samples = [sample for bucket in buckets.values() for sample in bucket]
             if cancelled():
                 break  # Do not call a partially decoded recording validated.
             if not samples:
@@ -123,6 +143,8 @@ def main():
             held_out = is_held_out(session_id)
             if held_out:
                 if candidate_schema != schema:
+                    dataset_stats.append({**stats, 'partition': 'heldOut', 'sampledWindows': len(samples),
+                                          'skipReason': 'Vocabulary absent from training data'})
                     continue  # Retry after training data introduces these command shapes.
                 validation.extend(samples)
             else:
@@ -136,15 +158,20 @@ def main():
             profile = metadata.get('controlProfile')
             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))
+            stats.update({'partition': 'heldOut' if held_out else 'training', 'sampledWindows': len(samples)})
+            dataset_stats.append(stats)
+            emit('dataset', sessionId=session_id, windows=seen, sampled=len(samples), diagnostics=stats)
         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():
+    if len(pool) < config['minimumSamples'] or not any(np.max(np.abs(sample[5])) > .02 for sample in pool) or cancelled():
         emit('complete', trained=False, accepted=[], rejected=rejected, samples=len(pool),
+             diagnostics={'dataset': dataset_stats, 'decodeSeconds': time.monotonic() - decode_started},
              reason='Stopped/time limit' if cancelled() else 'Waiting for enough training-partition samples with human actions')
         return
 
+    decode_seconds = time.monotonic() - decode_started
+    train_started = time.monotonic()
     numeric_mask = torch.zeros(SLOTS, VALUES)
     for slot, descriptor in enumerate(schema):
         numeric_mask[slot, :len(descriptor['fields'])] = 1
@@ -161,16 +188,17 @@ def main():
             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, states, targets, values, available, wheel_targets, previous_wheels = [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)
+            action_logits, predictions, wheel_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
+            wheel_loss = functional.smooth_l1_loss(wheel_predictions, wheel_targets)
+            loss = wheel_loss + event_loss + value_loss
             if not torch.isfinite(loss):
                 raise ValueError('Nonfinite training loss; checkpoint not saved')
             loss.backward()
@@ -180,13 +208,16 @@ def main():
             examples += len(batch)
             losses.append(loss.item())
             if steps % 10 == 0:
-                emit('progress', steps=steps, examples=examples, loss=loss.item())
+                emit('progress', steps=steps, examples=examples, loss=loss.item(),
+                     losses={'wheels': wheel_loss.item(), 'events': event_loss.item(), 'accessoryValues': value_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
 
+    training_seconds = time.monotonic() - train_started
+    evaluation_started = time.monotonic()
     distinct_windows.update(observed_windows)
     unique_count = sum(distinct_windows.values())
     evaluation = {'status': 'waiting for held-out sessions', 'windows': len(validation)}
@@ -195,13 +226,15 @@ def main():
         true_positive = predicted_positive = actual_positive = 0
         value_error = value_count = 0
         evaluated = 0
+        wheel_error = stopped_error = persistence_error = 0
+        changed_error = changed_persistence = changed_count = moving_count = 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)
+                images, states, targets, values, available, wheel_targets, previous_wheels = [torch.from_numpy(np.stack(items)) for items in zip(*batch)]
+                logits, predictions, wheel_predictions = model(images.float() / 255, states)
                 predicted = (logits.sigmoid() >= job['actionThreshold']) & available.bool()
                 actual = targets.bool() & available.bool()
                 true_positive += (predicted & actual).sum().item()
@@ -210,16 +243,42 @@ def main():
                 mask = numeric_mask.unsqueeze(0) * targets.unsqueeze(-1)
                 value_error += ((predictions - values).square() * mask).sum().item()
                 value_count += mask.sum().item()
+                errors = (wheel_predictions - wheel_targets).abs().mean(dim=1) * 500
+                persistence = (previous_wheels - wheel_targets).abs().mean(dim=1) * 500
+                changed = (previous_wheels - wheel_targets).abs().amax(dim=1) > .02
+                changed_error += errors[changed].sum().item()
+                changed_persistence += persistence[changed].sum().item()
+                changed_count += changed.sum().item()
+                moving_count += (wheel_targets.abs().amax(dim=1) > .02).sum().item()
+                wheel_error += errors.sum().item()
+                stopped_error += wheel_targets.abs().mean(dim=1).sum().item() * 500
+                persistence_error += persistence.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)
+        wheel_mae = wheel_error / max(1, evaluated)
+        stopped_mae = stopped_error / max(1, evaluated)
+        persistence_mae = persistence_error / max(1, evaluated)
+        changed_mae = changed_error / max(1, changed_count)
+        changed_baseline = changed_persistence / max(1, changed_count)
+        # Check control changes separately: copying the previous speed can look
+        # excellent on long straight runs without learning to steer or stop.
+        wheels_pass = (evaluated == len(validation) and evaluated >= 128 and moving_count >= 32
+                       and changed_count >= 32 and wheel_mae < stopped_mae
+                       and wheel_mae <= persistence_mae and changed_mae < changed_baseline)
+        accessories_pass = (predicted_positive == 0 if actual_positive == 0 else
+                            actual_positive >= 32 and precision >= .5 and recall >= .5)
+        passed = wheels_pass and accessories_pass and (value_count == 0 or 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}
+                      'numericMse': mse, 'positiveActions': actual_positive,
+                      'wheelMaeMmPerSecond': wheel_mae, 'stoppedBaselineMaeMmPerSecond': stopped_mae,
+                      'previousSpeedBaselineMaeMmPerSecond': persistence_mae,
+                      'changedWheelMaeMmPerSecond': changed_mae, 'changedBaselineMaeMmPerSecond': changed_baseline,
+                      'changedWindows': changed_count, 'movingWindows': moving_count,
+                      'wheelsPassed': wheels_pass, 'accessoriesPassed': accessories_pass}
+    evaluation_seconds = time.monotonic() - evaluation_started
     publication = {'distinctMinutes': unique_count / 600,
                    'newMinutes': (unique_count - published_windows) / 600,
                    'minimumMinutes': config['minimumDrivingMinutes'],
@@ -236,6 +295,7 @@ def main():
     if cancelled():
         reasons.append('Job time budget reached')
     publication['reason'] = '; '.join(reasons) or 'Published'
+    save_started = time.monotonic()
     published = None
     if not reasons:
         adjectives = ['amber', 'curious', 'gentle', 'quiet', 'silver', 'bright', 'merry', 'sleepy']
@@ -270,9 +330,15 @@ def main():
         'controlProfiles': control_profiles, 'distinctWindows': distinct_windows,
         'publishedWindows': published_windows,
     }, resume_path)
+    diagnostics = {'dataset': dataset_stats, 'decodeSeconds': decode_seconds,
+                   'trainingSeconds': training_seconds, 'evaluationSeconds': evaluation_seconds, 'checkpointSeconds': time.monotonic() - save_started,
+                   'cpuSeconds': time.process_time() - cpu_started,
+                   'peakRssMiB': resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024,
+                   'dependencies': {'torch': str(torch.__version__), 'numpy': str(np.__version__), 'av': str(av.__version__)},
+                   'policy': specification(schema), 'browserVideoLatencyMs': None}
     emit('complete', trained=True, accepted=accepted, rejected=rejected,
          steps=steps, examples=examples, loss=sum(losses) / len(losses),
-         published=published, publication=publication, evaluation=evaluation, elapsedSeconds=time.monotonic() - started)
+         published=published, publication=publication, evaluation=evaluation, diagnostics=diagnostics, elapsedSeconds=time.monotonic() - started)
 
 
 if __name__ == '__main__':