slop

server/src/services/roverLearningService/configuration.js

@@ -36,12 +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.' }),
-      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.' }),
+      minimumDrivingMinutes: number({ minimum: 1, maximum: 100000, description: 'Minimum distinct usable training footage in minutes before promoting Latest; repeated passes do not count.' }),
+      newDrivingMinutesPerModel: number({ minimum: 1, maximum: 100000, description: 'Additional distinct usable footage required between Latest promotions.' }),
       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.' }),
+      checkpointEverySteps: integer({ minimum: 1, maximum: 100000, description: 'Accumulated training steps between gated Latest promotions. Permanent named snapshots are created by administrators.' }),
       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', 'minimumDrivingMinutes', 'newDrivingMinutesPerModel', 'maxStepsPerJob', 'checkpointEverySteps', 'intervalSeconds', 'maxJobSeconds'] }),
+    }, { description: 'Ongoing CPU learning, sampling, and gated Latest checkpoint updates.', 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/diagnostics.js

@@ -6,7 +6,7 @@ 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']) {
+  'models.js', 'driving.js', 'recording.js', 'training.js', 'socketGateway.js', 'diagnostics.js']) {
   hash.update(fs.readFileSync(path.join(__dirname, file)));
 }
 const implementationSha256 = hash.digest('hex');
@@ -46,7 +46,7 @@ function report(state, controller) {
         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,
+    model: session ? { id: session.modelId, checkpointId: session.checkpointId, 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,

server/src/services/roverLearningService/driving.js

@@ -1,5 +1,4 @@
 // Autonomy owns a live user's control context, never an administrative identity.
-const fs = require('node:fs/promises');
 const path = require('node:path');
 const { spawn } = require('node:child_process');
 const roverManager = require('../roverManager');
@@ -7,6 +6,7 @@ const { isVerified, isDeterred } = require('../verificationService');
 const { isAdmin } = require('../roleService');
 const { submitCommand, issueCommand, subscribeCommandRecording } = require('../commandService');
 const { recordCommandRequest } = require('../commandService/recording');
+const { acquireModel } = require('./models');
 const { compatible } = require('./capabilities');
 
 const OPERATING = new Set(['drive', 'motors', 'servo', 'headlight', 'laser', 'horn', 'peripheral', 'song', 'raw']);
@@ -84,19 +84,19 @@ function createDriving({ root, config, onChange, activity }) {
 
   async function start(socket, roverId, modelId) {
     if (stopping || !canControl(socket, roverId)) throw new Error('You cannot control this rover');
-    if (!/^[0-9a-f-]{36}$/.test(modelId)) throw new Error('Invalid model');
-    const directory = path.join(root, 'models', modelId);
-    const metadata = JSON.parse(await fs.readFile(path.join(directory, 'model.json'), 'utf8'));
-    if (metadata.id !== modelId || !compatible(metadata, roverManager.rovers.get(roverId))) {
-      throw new Error('Model is not compatible with this rover');
-    }
-    if (stopping || !canControl(socket, roverId)) throw new Error('Control permission changed');
-    if (sessions.has(roverId)) throw new Error('Stop the current controller before starting another model');
+    const acquired = await acquireModel(root, modelId);
+    const { directory, metadata } = acquired;
+    try {
+      if (!compatible(metadata, roverManager.rovers.get(roverId))) throw new Error('Model is not compatible with this rover');
+      if (stopping || !canControl(socket, roverId)) throw new Error('Control permission changed');
+      if (sessions.has(roverId)) throw new Error('Stop the current controller before starting another model');
+    } catch (error) { await acquired.release(); throw error; }
     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, threshold: config.driving.actionThreshold, history: [] };
+    const session = { worker, socket, roverId, modelId, checkpointId: metadata.id, modelName: modelId === 'latest' ? 'Latest' : modelId === 'previous' ? 'Previous' : metadata.name, status: 'starting', startedAt: Date.now(), lastPredictionAt: null, threshold: config.driving.actionThreshold, history: [] };
     session.closed = new Promise((resolve) => worker.once('close', resolve));
+    session.closed.then(() => acquired.release()).catch((error) => activity('Checkpoint release failed', { error: error.message }));
     sessions.set(roverId, session);
     recent.delete(roverId);
     let output = '';

server/src/services/roverLearningService/models.js

@@ -1,20 +1,119 @@
-// Published directories are immutable and become visible via atomic rename.
+// Mutable aliases reference immutable checkpoints. All readers and retention share
+// a queue so a model cannot disappear between resolving its alias and loading it.
 const fs = require('node:fs/promises');
 const path = require('node:path');
-
-async function listModels(root) {
-  const directory = path.join(root, 'models');
-  let entries;
-  try { entries = await fs.readdir(directory, { withFileTypes: true }); }
-  catch (error) { if (error.code === 'ENOENT') return []; throw error; }
-  const models = [];
-  for (const entry of entries) {
-    if (!entry.isDirectory() || !/^[0-9a-f-]{36}$/.test(entry.name)) continue;
-    const metadata = JSON.parse(await fs.readFile(path.join(directory, entry.name, 'model.json'), 'utf8'));
-    if (metadata.id !== entry.name) throw new Error(`Model identity mismatch: ${entry.name}`);
-    models.push(metadata);
-  }
-  return models.sort((a, b) => b.createdAt - a.createdAt);
+const { randomUUID, randomInt } = require('node:crypto');
+const UUID = /^[0-9a-f-]{36}$/;
+let queue = Promise.resolve();
+const leases = new Map();
+const retired = new Set();
+function serial(task) {
+  const result = queue.then(task);
+  queue = result.catch(() => {});
+  return result;
 }
-
-module.exports = { listModels };
+const checkpointRoot = (root) => path.join(root, 'training', 'checkpoints');
+async function index(root) {
+  try { return JSON.parse(await fs.readFile(path.join(checkpointRoot(root), 'index.json'), 'utf8')); }
+  catch (error) { if (error.code === 'ENOENT') return {}; throw error; }
+}
+async function metadata(directory) {
+  const model = JSON.parse(await fs.readFile(path.join(directory, 'model.json'), 'utf8'));
+  if (model.id !== path.basename(directory)) throw new Error('Model identity mismatch');
+  return model;
+}
+async function prune(root, state) {
+  const directory = checkpointRoot(root);
+  await fs.mkdir(directory, { recursive: true });
+  for (const entry of await fs.readdir(directory, { withFileTypes: true })) {
+    if (!entry.isDirectory()) continue;
+    if (/^\.[0-9a-f-]{36}\.pending$/.test(entry.name)) {
+      await fs.rm(path.join(directory, entry.name), { recursive: true });
+      continue;
+    }
+    if (!UUID.test(entry.name)) continue;
+    const target = path.join(directory, entry.name);
+    if (!Object.values(state).includes(entry.name)) {
+      if (leases.has(target)) retired.add(target);
+      else { await fs.rm(target, { recursive: true }); retired.delete(target); }
+    }
+  }
+}
+function adoptCheckpoint(root, model, promote) {
+  return serial(async () => {
+    const state = await index(root);
+    if (!UUID.test(model.id)) throw new Error('Invalid checkpoint identity');
+    await metadata(path.join(checkpointRoot(root), model.id));
+    state.candidate = model.id;
+    if (promote) { state.previous = state.latest || null; state.latest = model.id; }
+    const filename = path.join(checkpointRoot(root), 'index.json');
+    await fs.writeFile(`${filename}.tmp`, JSON.stringify(state));
+    await fs.rename(`${filename}.tmp`, filename);
+    await prune(root, state);
+  });
+}
+function acquireModel(root, id) {
+  return serial(async () => {
+    const state = await index(root);
+    const alias = ['latest', 'previous'].includes(id);
+    const resolved = alias ? state[id] : id;
+    if (!UUID.test(resolved || '')) throw new Error('Model is not available');
+    const directory = path.join(alias ? checkpointRoot(root) : path.join(root, 'models'), resolved);
+    const model = await metadata(directory);
+    leases.set(directory, (leases.get(directory) || 0) + 1);
+    let released = false;
+    return { directory, metadata: model, release: () => serial(async () => {
+      if (released) return;
+      released = true;
+      const count = leases.get(directory) - 1;
+      if (count) leases.set(directory, count); else leases.delete(directory);
+      // Delete only a specific checkpoint already retired by the trainer, not
+      // a worker's new export awaiting adoption.
+      if (!count && retired.has(directory)) {
+        await fs.rm(directory, { recursive: true, force: true });
+        retired.delete(directory);
+      }
+    }) };
+  });
+}
+function saveSnapshot(root) {
+  return serial(async () => {
+    const state = await index(root);
+    if (!UUID.test(state.candidate || '')) throw new Error('No completed learner checkpoint yet');
+    const source = path.join(checkpointRoot(root), state.candidate);
+    const original = await metadata(source);
+    const id = randomUUID();
+    const words = [['amber', 'silver', 'quiet', 'curious'], ['otter', 'finch', 'fox', 'wren']];
+    const name = `${words[0][randomInt(4)]}-${words[1][randomInt(4)]}-${id.slice(0, 8)}`;
+    const model = { ...original, id, name, createdAt: Date.now(), sourceCheckpointId: original.id,
+      snapshot: true, experimental: original.evaluation !== 'passed imitation gate' };
+    const destination = path.join(root, 'models', id);
+    const pending = path.join(root, 'models', `.${id}.pending`);
+    await fs.mkdir(pending, { recursive: true });
+    try {
+      await fs.copyFile(path.join(source, 'weights.pt'), path.join(pending, 'weights.pt'));
+      await fs.writeFile(path.join(pending, 'model.json'), JSON.stringify(model));
+      await fs.rename(pending, destination);
+    } catch (error) { await fs.rm(pending, { recursive: true, force: true }); throw error; }
+    return model;
+  });
+}
+function listModels(root) {
+  return serial(async () => {
+    let entries;
+    try { entries = await fs.readdir(path.join(root, 'models'), { withFileTypes: true }); }
+    catch (error) { if (error.code !== 'ENOENT') throw error; entries = []; }
+    const models = [];
+    const state = await index(root);
+    for (const alias of ['latest', 'previous']) {
+      if (!state[alias]) continue;
+      const model = await metadata(path.join(checkpointRoot(root), state[alias]));
+      models.push({ ...model, id: alias, checkpointId: model.id, name: alias === 'latest' ? 'Latest' : 'Previous', automatic: true });
+    }
+    for (const entry of entries) {
+      if (entry.isDirectory() && UUID.test(entry.name)) models.push(await metadata(path.join(root, 'models', entry.name)));
+    }
+    return models;
+  });
+}
+module.exports = { listModels, acquireModel, adoptCheckpoint, saveSnapshot };

server/src/services/roverLearningService/socketGateway.js

@@ -2,7 +2,8 @@
 const io = require('../../globals/io');
 const roverManager = require('../roverManager');
 const { isVerified } = require('../verificationService');
-const { listModels } = require('./models');
+const { isAdmin } = require('../roleService');
+const { listModels, saveSnapshot } = require('./models');
 const { report } = require('./diagnostics');
 const { compatible } = require('./capabilities');
 
@@ -25,11 +26,11 @@ function createGateway({ root, driving, getSystemState, activity }) {
     const system = getSystemState();
     const state = {
       available: true, roverId: id, updatedAt: Date.now(),
-      control: { canStart: canControl && !session, canAdjust: canControl && Boolean(session), canStop: Boolean(session && (canControl || session.socketId === socket.id)),
+      control: { canSnapshot: isAdmin(socket), 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,
+        .map((model) => ({ id: model.id, checkpointId: model.checkpointId || model.id, automatic: Boolean(model.automatic), experimental: Boolean(model.experimental), name: model.name, createdAt: model.createdAt,
           trainingSteps: model.trainingSteps, examplesProcessed: model.examplesProcessed,
           weightBytes: model.weightBytes, parameters: model.parameters, trainingLoss: model.trainingLoss,
           formatVersion: model.specification?.version, evaluation: model.evaluation, validation: model.validation, distinctMinutes: model.distinctMinutes, controls: model.specification?.commands?.length || 0,
@@ -76,7 +77,13 @@ function createGateway({ root, driving, getSystemState, activity }) {
       try {
         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 || ''));
+        if (payload.action === 'snapshot') {
+          if (!isAdmin(socket)) throw new Error('Administrator access required');
+          const model = await saveSnapshot(root);
+          models = await listModels(root);
+          lastCatalog = Date.now();
+          activity('Learner snapshot saved', { model: model.name });
+        } else 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);

server/src/services/roverLearningService/training.js

@@ -4,6 +4,7 @@ const fs = require('node:fs/promises');
 const path = require('node:path');
 const { spawn } = require('node:child_process');
 const { randomInt } = require('node:crypto');
+const { adoptCheckpoint } = require('./models');
 const { listRecordings } = require('./storage');
 
 function createTrainer({ root, config, actionThreshold, logger }) {
@@ -133,6 +134,7 @@ function createTrainer({ root, config, actionThreshold, logger }) {
             if (!stopped) logger.error('Training job failed', { error: state.error, stderr });
           } else {
             for (const [id, reason] of Object.entries(result.rejected || {})) ledger[id] = { rejected: reason };
+            if (result.published) await adoptCheckpoint(root, result.published, result.promote);
             if (result.trained) {
               for (const id of result.accepted) ledger[id] = { rounds: (ledger[id]?.rounds || 0) + 1 };
             }

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

@@ -294,33 +294,36 @@ def main():
         reasons.append(evaluation['status'])
     if cancelled():
         reasons.append('Job time budget reached')
-    publication['reason'] = '; '.join(reasons) or 'Published'
+    publication['reason'] = '; '.join(reasons) or 'Latest updated'
     save_started = time.monotonic()
     published = None
+    # Every completed job exports an immutable candidate; Node owns promotion and retention.
+    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]}'
+    checkpoint_root = root / 'training' / 'checkpoints'
+    checkpoint_root.mkdir(exist_ok=True)
+    staging = checkpoint_root / 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': 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(checkpoint_root / model_id)
     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())
-        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': 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, {
@@ -338,7 +341,7 @@ def main():
                    '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, diagnostics=diagnostics, elapsedSeconds=time.monotonic() - started)
+         published=published, promote=not reasons, publication=publication, evaluation=evaluation, diagnostics=diagnostics, elapsedSeconds=time.monotonic() - started)
 
 
 if __name__ == '__main__':