diff --git a/plans/rover learning service.md b/plans/rover learning service.md index f171fe4c..730f193c 100644 --- a/plans/rover learning service.md +++ b/plans/rover learning service.md @@ -24,6 +24,10 @@ The user declined the manual server checklist and explicitly asked to continue. Do not block further work or repeatedly ask them to do it. Preserve the unverified status and have the reader reject unusable recordings automatically. +Latest increment: internal card gating, dock lead-in/tail capture, live prediction +diagnostics, ongoing retained-data sampling, and gated model publication are coded. +See the final handoff section for settings and the new training lineage. + Next work: verify the integrated feature against actual recordings and the real camera/rover pipeline when that environment is available; inspect the VIP card in the running browser. Do not invent successful driving results or block further @@ -105,7 +109,7 @@ the default development path. Intended layout: rover-learning/ recordings/ Rolling video/events/session metadata (implemented) models/ Permanent UUID directories: weights.pt and model.json - training/ resume.pt, schedule.json, and current job.json + training/ resume-v2.pt, schedule.json, and current job.json ``` Training/model directories are created only when service and training are enabled. @@ -533,9 +537,9 @@ Training defaults beneath `roverLearning.training`: | batchSize | 16 | | maxSamples / minimumSamples | 1024 / 128 | | passesPerJob | 2 | -| roundsPerSession | 3 | +| minimumDrivingMinutes / newDrivingMinutesPerModel | 30 / 10 | | maxStepsPerJob | 200 | -| checkpointEverySteps | 100 | +| checkpointEverySteps | 1000 | | intervalSeconds | 60 | | maxJobSeconds | 600 | @@ -548,9 +552,9 @@ override these defaults; this change does not rewrite deployed configuration. At most four completed retained sessions are selected per job. Samples are spread over each full video via reservoir sampling, not always taken from its beginning. -The rounds limit is sampled training visits, not full passes over all recorded -frames. Once retained data reaches that limit, training waits for new sessions. -Candidate publication happens at job boundaries after enough accumulated steps; +The prior three-visit limit is removed. Retained usable training sessions remain +eligible for repeated sampling; finite jobs still yield between runs. +Candidate publication happens at job boundaries after data, step, and held-out gates; it does not automatically deploy a model. No all-no-action sample pool is trained. Container rebuild is needed to include Python packages. For a non-container @@ -616,3 +620,57 @@ record verification limits and outstanding server/hardware work explicitly. end-to-end inference/training and browser rendering remain unverified. No new tests or background processes were added for verification, and no full build or full test suite was run. + + +### Dock capture, live diagnostics, and publication correction + +Authorized by IPI after the user reported idle dock recording, many premature +models, silent controllers, and training stopping after a few jobs. + +- VipPanel now renders only ``; the card owns feature and + verification gating, rover selection context, and grid wrapper. The user's + pre-existing card edit was preserved. +- Human docked sessions maintain disposable packet-copy video segments and a + bounded five-second event buffer. Segments are keyframe-aligned (target two + seconds); old closed segments are pruned, with an extra edge segment retained + for decoding. Undocking promotes the buffer and sets trainingStartedAt to five + seconds earlier. After docking, recording closes after five seconds (250ms + polling granularity) and returns to buffering. Never-undocked buffers are + deleted on close/recovery and never offered to training. Disk retention still + applies. The reader handles chronological segments and filters the lead-in + boundary; no RTSP restart is needed to promote a dock buffer. +- The service still excludes autonomous sessions and still requires an eligible + human. Human input does not stop or pause autonomy. +- Expanded card diagnostics opt into complete volatile snapshots at 10Hz; + collapsed cards receive 1Hz. Predictions include all known command shapes, + scores, decoded controls, below-threshold/competing/latched reasons, and actual + submission results. "Predicting no new action" is separate from worker running. +- Removed roundsPerSession from configuration and scheduling. Completed usable + recordings remain eligible indefinitely within rolling retention. Jobs remain + finite, lower-priority, and separated by the configured interval. Waiting and + publication reasons are visible in the card. +- Default publication gates: 1000 optimizer steps, 30 distinct usable training + minutes, and 10 additional distinct minutes since the previous publication. + Usable minutes count enumerated aligned 100ms windows once per training session, + not repeated optimizer exposures; only a reservoir is used in each job. +- Entire sessions are deterministically partitioned 80/20 by session-ID hash. + Held-out sessions never supply gradient updates or new command vocabulary. + Evaluation requires at least 128 windows and 32 positive events, precision and + recall >=0.5 at the configured action threshold, and normalized numeric MSE + <=0.1 with numeric targets present. These are initial imitation gates, not proof + of autonomous competence. Evaluation uses the current job's held-out reservoir; + jobs with no usable held-out cohort cannot publish. +- Fresh `training/resume-v2.pt` prevents old training exposure contaminating the + new validation split. Existing published models and old resume.pt are preserved. + New checkpoints retain optimizer progress even when publication is blocked. +- Deployment config: remove the old `training.roundsPerSession` key because the + schema is strict. Set `checkpointEverySteps: 1000`, `minimumDrivingMinutes: 30`, + and `newDrivingMinutesPerModel: 10`; saved values override changed defaults. +- Verification: four focused configuration tests passed, modified Node entry + points passed syntax checks, targeted VIP ESLint passed, and the production + training dependency entry point loaded the policy and segmented dataset reader. + No fixtures/test files, full builds, or background servers were created. +- Still unverified on this development machine: live RTSP segment timestamps and + dock transitions, actual optimizer/evaluation/publication runs on recordings, + browser appearance/live updates, and real autonomous behavior. Do not claim + these checks passed or that the existing event-based policy now drives well. diff --git a/server/src/services/roverLearningService/configuration.js b/server/src/services/roverLearningService/configuration.js index b865d89c..762cc5bd 100644 --- a/server/src/services/roverLearningService/configuration.js +++ b/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'] }), }; diff --git a/server/src/services/roverLearningService/driving.js b/server/src/services/roverLearningService/driving.js index e4302da0..5825d3b2 100644 --- a/server/src/services/roverLearningService/driving.js +++ b/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; } diff --git a/server/src/services/roverLearningService/recording.js b/server/src/services/roverLearningService/recording.js index 45c4caaa..c94d084f 100644 --- a/server/src/services/roverLearningService/recording.js +++ b/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'); }, }; diff --git a/server/src/services/roverLearningService/runtime.js b/server/src/services/roverLearningService/runtime.js index aca2a996..c7580943 100644 --- a/server/src/services/roverLearningService/runtime.js +++ b/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(); diff --git a/server/src/services/roverLearningService/socketGateway.js b/server/src/services/roverLearningService/socketGateway.js index 393edb14..4c82b1c0 100644 --- a/server/src/services/roverLearningService/socketGateway.js +++ b/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() { diff --git a/server/src/services/roverLearningService/storage.js b/server/src/services/roverLearningService/storage.js index 7fd306db..288b36df 100644 --- a/server/src/services/roverLearningService/storage.js +++ b/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', diff --git a/server/src/services/roverLearningService/training.js b/server/src/services/roverLearningService/training.js index 7280518d..465874c6 100644 --- a/server/src/services/roverLearningService/training.js +++ b/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 }); } diff --git a/server/src/services/roverLearningService/workers/dataset.py b/server/src/services/roverLearningService/workers/dataset.py index 40cfeb5e..61516197 100644 --- a/server/src/services/roverLearningService/workers/dataset.py +++ b/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 diff --git a/server/src/services/roverLearningService/workers/infer.py b/server/src/services/roverLearningService/workers/infer.py index 9013108d..a0d41239 100644 --- a/server/src/services/roverLearningService/workers/infer.py +++ b/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__': diff --git a/server/src/services/roverLearningService/workers/train.py b/server/src/services/roverLearningService/workers/train.py index 70cc1b42..4c004de8 100644 --- a/server/src/services/roverLearningService/workers/train.py +++ b/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__': diff --git a/webui/src/components/VipPanel/index.jsx b/webui/src/components/VipPanel/index.jsx index ea3ff01e..532f1888 100644 --- a/webui/src/components/VipPanel/index.jsx +++ b/webui/src/components/VipPanel/index.jsx @@ -55,9 +55,6 @@ export default function VipPanel() { return (
- {isVerified && session?.features?.roverLearning ? ( -
- ) : null} {!isVerified ? (
{isVerified ? (
+ ; } -export default function VipRoverLearningCard({ roverId }) { +export default function VipRoverLearningCard() { + const enabled = useSessionSelector((state) => isFeatureEnabled(state, 'roverLearning') && Boolean(state.session?.isVerified)); + const roverId = useSessionSelector((state) => state.session?.assignment?.roverId || null); + if (!enabled) return null; + return ; +} + +function LearningCard({ roverId }) { const socket = useSocket(); const [snapshot, setSnapshot] = useState(null); const [selectedId, setSelectedId] = useState(''); + const [detailsOpen, setDetailsOpen] = useState(false); const [working, setWorking] = useState(false); const [error, setError] = useState(''); const { value: settings, save } = useSettingsNamespace('roverLearning', { favorites: [] }); @@ -37,7 +47,7 @@ export default function VipRoverLearningCard({ roverId }) { const receive = (state) => { if (active) setSnapshot(state); }; const subscribe = () => { if (!socket.connected) return; - socket.timeout(5000).emit('roverLearning:subscribe', { roverId }, (failure, response) => { + socket.timeout(5000).emit('roverLearning:subscribe', { roverId, live: detailsOpen }, (failure, response) => { if (!failure && response?.state) receive(response.state); }); }; @@ -57,7 +67,7 @@ export default function VipRoverLearningCard({ roverId }) { socket.off('disconnect', disconnected); socket.emit('roverLearning:unsubscribe'); }; - }, [socket, roverId]); + }, [socket, roverId, detailsOpen]); const state = snapshot?.roverId === roverId ? snapshot : null; const models = [...(state?.models || [])].sort((a, b) => Number(favorites.includes(b.id)) - Number(favorites.includes(a.id)) || b.createdAt - a.createdAt); @@ -104,10 +114,10 @@ export default function VipRoverLearningCard({ roverId }) {
{state?.control?.reason &&

{state.control.reason}

} -

Manual controls stay available and do not stop the model.

-

Recording: {recording ? recording.paused ? 'paused for disk space' : `${recording.sessions.length} human session(s)` : 'unavailable'} · Training: {training?.status || 'unavailable'}

+ {/*

Manual controls stay available and do not stop the model.

*/} +

Recording: {recording ? recording.paused ? 'paused for disk space' : `${recording.sessions.filter((item) => item.video === 'Recording human driving').length} driving · ${recording.sessions.filter((item) => item.video === 'Dock lead-in buffer').length} dock buffer(s)` : 'unavailable'} · Training: {training?.status || 'unavailable'}

{(error || state?.catalogError) &&

{error || state.catalogError}

} -
+
setDetailsOpen(event.currentTarget.open)}> System details
@@ -134,6 +144,11 @@ export default function VipRoverLearningCard({ roverId }) { {training?.lastAvailableWindows == null ? 'Unavailable' : `${training.lastAvailableWindows} / ${training.lastSampledWindows}`} {limits ? `${limits.training.threads} / ${limits.training.batchSize}` : 'Unavailable'} {limits ? `${limits.training.maxSamples} / ${limits.training.passesPerJob}` : 'Unavailable'} + {training?.publication?.distinctMinutes?.toFixed(1) ?? 'Unavailable'} + {limits ? `${limits.training.minimumDrivingMinutes} / ${limits.training.newDrivingMinutesPerModel}` : 'Unavailable'} + {training?.publication?.reason || 'Collecting data'} + {training?.evaluation?.status || 'Waiting for held-out data'} + {training?.evaluation?.precision == null ? 'Unavailable' : `${training.evaluation.precision.toFixed(3)} / ${training.evaluation.recall.toFixed(3)}`} {display(limits?.training.checkpointEverySteps)}
{training?.error &&

{training.error}

} {training?.reason &&

{training.reason}

} @@ -147,9 +162,13 @@ export default function VipRoverLearningCard({ roverId }) { {chosen ? `${chosen.trainingSteps} / ${chosen.examplesProcessed}` : 'Unavailable'} {display(chosen?.controls)} {chosen?.evaluation || 'Unavailable'} + {chosen?.distinctMinutes?.toFixed(1) ?? 'Unavailable'} + {chosen?.validation?.precision == null ? 'Unavailable' : `${chosen.validation.precision.toFixed(3)} / ${chosen.validation.recall.toFixed(3)}`} {models.length} -

Models are kept permanently. Favorites are saved in this browser.

 +
+ {session?.predictionStatus || 'Unavailable'} + {display(session?.threshold)} {session?.status || 'Stopped'} {session?.startedBy || 'Unavailable'} {age(now, session?.startedAt)} @@ -159,6 +178,13 @@ export default function VipRoverLearningCard({ roverId }) { {session?.lastResult || 'Unavailable'} {display(limits?.driving.threads)}
{session?.lastAction && 
{JSON.stringify(session.lastAction.command, null, 2)}
} + {session?.proposals?.length > 0 &&
+ {session.proposals.map((proposal) =>
+

{proposal.command.type} · score {proposal.score.toFixed(3)} · {proposal.reason}

+ 
{JSON.stringify(proposal.command)}
+
)} +
} + {session?.submissions?.map((item, index) =>

Submitted {item.command.type}: {item.result}

)} {session?.stopReason &&

{session.stopReason}

}