micrationslop

perf/nomad-benchmark.py

@@ -0,0 +1,208 @@
+#!/usr/bin/env python3
+"""CPU benchmark of published NoMaD exploration, without ROS or rover connections.
+
+Run: python3 perf/nomad-benchmark.py --setup
+Subsequent runs: python3 perf/nomad-benchmark.py
+Quick run: python3 perf/nomad-benchmark.py --threads 4 --samples 1 --seconds 2 --iterations 2
+
+--setup installs CPU dependencies in a private cached venv and downloads pinned
+upstream source plus the official checkpoint. Nothing touches server packages.
+Synthetic images measure compute, NOT navigation quality. Excludes capture,
+video decode, network, physical control, training, and concurrent rover load.
+"""
+import argparse
+import hashlib
+import importlib.metadata
+import json
+import math
+import os
+from pathlib import Path
+import platform
+import resource
+import statistics
+import subprocess
+import sys
+import time
+import urllib.request
+import venv
+
+VINT = 'dca79815b704e5aa9c6bdc3082351f9e3b2848c2'
+DIFFUSION = '5ba07ac6661db573af695b419a7947ecb704690f'
+CHECKPOINT_ID = '1YJhkkMJAYOiKNyCaelbS_alpUpAJsOUb'
+SOURCES = {
+    ('robodhruv/visualnav-transformer', VINT): [
+        'train/config/nomad.yaml', 'train/vint_train/data/data_config.yaml',
+        'train/vint_train/models/nomad/nomad.py',
+        'train/vint_train/models/nomad/nomad_vint.py',
+        'train/vint_train/models/vint/self_attention.py',
+    ],
+    ('real-stanford/diffusion_policy', DIFFUSION): [
+        'diffusion_policy/model/diffusion/conditional_unet1d.py',
+        'diffusion_policy/model/diffusion/conv1d_components.py',
+        'diffusion_policy/model/diffusion/positional_embedding.py',
+    ],
+}
+
+
+def positive(value):
+    result = int(value)
+    if result < 1:
+        raise argparse.ArgumentTypeError('Must be positive')
+    return result
+
+
+def arguments():
+    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument('--setup', action='store_true')
+    p.add_argument('--cache', type=Path, default=Path.home() / '.cache/rover-nomad-benchmark')
+    p.add_argument('--threads', type=positive, nargs='+', default=[4, 8, 16])
+    p.add_argument('--samples', type=positive, nargs='+', default=[1, 8], help='Candidate trajectories per decision, not rover count')
+    p.add_argument('--seconds', type=positive, default=10)
+    p.add_argument('--iterations', type=positive, default=10)
+    p.add_argument('--worker', action='store_true', help=argparse.SUPPRESS)
+    p.add_argument('--result', type=Path, help=argparse.SUPPRESS)
+    return p.parse_args()
+
+
+def download(url, path):
+    if path.exists():
+        return
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + '.part')
+    print('Downloading', path.name, flush=True)
+    with urllib.request.urlopen(url, timeout=120) as response, temporary.open('wb') as output:
+        import shutil
+        shutil.copyfileobj(response, output)
+    temporary.replace(path)
+
+
+def prepare(args):
+    python = args.cache / 'venv/bin/python'
+    if not python.exists():
+        venv.EnvBuilder(with_pip=True).create(args.cache / 'venv')
+    subprocess.run([str(python), '-m', 'pip', 'install', 'torch', 'torchvision',
+                    '--index-url', 'https://download.pytorch.org/whl/cpu'], check=True)
+    subprocess.run([str(python), '-m', 'pip', 'install', 'numpy', 'diffusers',
+                    'efficientnet-pytorch', 'einops', 'PyYAML', 'gdown'], check=True)
+    for (repo, revision), paths in SOURCES.items():
+        for path in paths:
+            download(f'https://raw.githubusercontent.com/{repo}/{revision}/{path}', args.cache / 'source' / revision / path)
+    checkpoint = args.cache / 'nomad.pth'
+    if not checkpoint.exists():
+        temporary = checkpoint.with_suffix('.part')
+        subprocess.run([str(python), '-m', 'gdown', CHECKPOINT_ID, '-O', str(temporary)], check=True)
+        temporary.replace(checkpoint)
+
+
+def worker(args):
+    import numpy as np
+    import torch
+    import yaml
+    from diffusers.schedulers.scheduling_ddpm import DDPMScheduler
+    sys.path[:0] = [str(args.cache / 'source' / VINT / 'train'), str(args.cache / 'source' / DIFFUSION)]
+    from vint_train.models.nomad.nomad import NoMaD, DenseNetwork
+    from vint_train.models.nomad.nomad_vint import NoMaD_ViNT, replace_bn_with_gn
+    from diffusion_policy.model.diffusion.conditional_unet1d import ConditionalUnet1D
+
+    threads, samples = args.threads[0], args.samples[0]
+    torch.set_num_threads(threads)
+    torch.set_num_interop_threads(1)
+    torch.manual_seed(0)
+    c = yaml.safe_load((args.cache / 'source' / VINT / 'train/config/nomad.yaml').read_text())
+    vision = replace_bn_with_gn(NoMaD_ViNT(
+        context_size=c['context_size'], obs_encoding_size=c['encoding_size'],
+        mha_num_attention_heads=c['mha_num_attention_heads'],
+        mha_num_attention_layers=c['mha_num_attention_layers'], mha_ff_dim_factor=c['mha_ff_dim_factor']))
+    model = NoMaD(vision, ConditionalUnet1D(input_dim=2, global_cond_dim=c['encoding_size'],
+        down_dims=c['down_dims'], cond_predict_scale=c['cond_predict_scale']), DenseNetwork(c['encoding_size'])).eval()
+    # Strict loading prevents accidentally benchmarking a partly random model.
+    model.load_state_dict(torch.load(args.cache / 'nomad.pth', map_location='cpu', weights_only=True), strict=True)
+    scheduler = DDPMScheduler(num_train_timesteps=c['num_diffusion_iters'],
+        beta_schedule='squaredcos_cap_v2', clip_sample=True, prediction_type='epsilon')
+    width, height = c['image_size']
+    frames = torch.rand(c['context_size'] + 1, 3, height, width)
+    mean = torch.tensor([.485, .456, .406]).view(1, 3, 1, 1)
+    std = torch.tensor([.229, .224, .225]).view(1, 3, 1, 1)
+    mask = torch.ones(1, dtype=torch.long)
+    stats = yaml.safe_load((args.cache / 'source' / VINT / 'train/vint_train/data/data_config.yaml').read_text())['action_stats']
+    low, high = np.array(stats['min']), np.array(stats['max'])
+
+    def decision():
+        start = time.perf_counter()
+        obs = ((frames - mean) / std).reshape(1, -1, height, width)
+        goal = torch.randn(1, 3, height, width)
+        cond = model('vision_encoder', obs_img=obs, goal_img=goal, input_goal_mask=mask).repeat(samples, 1)
+        encoded = time.perf_counter()
+        action = torch.randn(samples, c['len_traj_pred'], 2)
+        scheduler.set_timesteps(c['num_diffusion_iters'])
+        for step in scheduler.timesteps:
+            noise = model('noise_pred_net', sample=action, timestep=step, global_cond=cond)
+            action = scheduler.step(noise, step, action).prev_sample
+        # Match upstream get_action: unnormalize deltas then accumulate waypoints.
+        waypoints = np.cumsum((action.numpy() + 1) / 2 * (high - low) + low, axis=1)
+        if not np.isfinite(waypoints).all():
+            raise RuntimeError('Non-finite model output')
+        end = time.perf_counter()
+        return (end - start) * 1000, (encoded - start) * 1000, (end - encoded) * 1000
+
+    rows = []
+    with torch.inference_mode():
+        for _ in range(3):
+            decision()
+        start, cpu = time.perf_counter(), time.process_time()
+        while len(rows) < args.iterations or time.perf_counter() - start < args.seconds:
+            rows.append(decision())
+        elapsed, cpu = time.perf_counter() - start, time.process_time() - cpu
+    total = sorted(r[0] for r in rows)
+    result = dict(threads=threads, candidateTrajectories=samples, iterations=len(rows),
+        p50Ms=statistics.median(total), p95Ms=total[math.ceil(.95 * len(total)) - 1], maxMs=max(total),
+        decisionsPerSecond=len(rows)/elapsed, cpuCores=cpu/elapsed,
+        peakMiB=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/1024,
+        visionMeanMs=statistics.mean(r[1] for r in rows), diffusionAndDecodeMeanMs=statistics.mean(r[2] for r in rows),
+        parameters=sum(p.numel() for p in model.parameters()), diffusionSteps=c['num_diffusion_iters'],
+        imageSize=c['image_size'], frames=c['context_size']+1,
+        versions={name: importlib.metadata.version(name) for name in ['torch','torchvision','diffusers','numpy','efficientnet-pytorch','einops']})
+    args.result.write_text(json.dumps(result))
+
+
+def main():
+    args = arguments()
+    args.cache = args.cache.expanduser().resolve()
+    if args.worker:
+        worker(args)
+        return
+    if args.setup:
+        prepare(args)
+    python = args.cache / 'venv/bin/python'
+    if not python.exists() or not (args.cache / 'nomad.pth').exists():
+        raise RuntimeError('First run: python3 perf/nomad-benchmark.py --setup')
+    print('CPU FP32 NoMaD exploration; synthetic images; no driving-quality measurement.', flush=True)
+    print('Candidates are trajectories for ONE rover. Includes vision + all 10 diffusion steps.', flush=True)
+    print('threads candidates  p50 ms  p95 ms  max ms decisions/s CPU cores peak MiB', flush=True)
+    results = []
+    import tempfile
+    with tempfile.TemporaryDirectory(prefix='nomad-bench-') as directory:
+        for threads in args.threads:
+            for samples in args.samples:
+                path = Path(directory) / 'result.json'
+                subprocess.run([str(python), str(Path(__file__).resolve()), '--worker', '--cache', str(args.cache),
+                    '--threads', str(threads), '--samples', str(samples), '--seconds', str(args.seconds),
+                    '--iterations', str(args.iterations), '--result', str(path)], check=True,
+                    timeout=max(600, args.seconds*20), env={**os.environ, 'PYTHONDONTWRITEBYTECODE':'1'})
+                r = json.loads(path.read_text()); results.append(r)
+                print(f"{threads:7} {samples:10} {r['p50Ms']:7.1f} {r['p95Ms']:7.1f} {r['maxMs']:7.1f} {r['decisionsPerSecond']:11.2f} {r['cpuCores']:9.2f} {r['peakMiB']:8.0f}", flush=True)
+    with (args.cache / 'nomad.pth').open('rb') as f:
+        digest = hashlib.file_digest(f, 'sha256').hexdigest()
+    report = dict(benchmark='nomad-cpu-v1', platform=platform.platform(), logicalCpus=os.cpu_count(),
+        loadAverage=os.getloadavg(), sourceRevisions=dict(vint=VINT,diffusion=DIFFUSION), checkpointSha256=digest,
+        exclusions=['capture','video decode','network','physical control','training','concurrent model workloads'], results=results)
+    print('\nPaste this report back:\n' + json.dumps(report, separators=(',', ':')))
+    print('At 10 decisions/s the total budget is 100ms; at 5/s it is 200ms. Leave room for the excluded pipeline.')
+
+
+if __name__ == '__main__':
+    try:
+        main()
+    except (Exception, KeyboardInterrupt) as error:
+        print(f'Benchmark failed: {error}', file=sys.stderr)
+        sys.exit(1)

plans/rover learning service.md

@@ -2,6 +2,19 @@
 
 ## Resume here
 
+Latest repair: full learning-path audit. Fixed missing command labels between
+video samples with a shared causal 100ms grid; added real-data score/baseline/
+training-probe diagnostics and normalized saved control profiles for retained
+recordings. Continues policy 5 / lineage v6; this audit does not reset progress.
+See the final audit results section for measured synthetic evidence and limits.
+
+Latest scope (authorized IPI): only drive, camera tilt, headlight and the four
+operating-mode/dock commands (128/131/132/143). Exclude horn, song, laser and all
+brush/vacuum motor commands from both learning and execution, including raw OI
+bypasses. Raw recordings stay intact. Policy format 5, training lineage v6;
+startup removes earlier models and lineages through v5. Earlier scope below is
+historical. No configuration change required.
+
 Latest change (2026-09-29 IPI): format 4 seeds known user-control command shapes
 before training. Generic peripherals are excluded; sensorStream is ignored without
 invalidating footage. Drive includes zero-speed drive, with no separate stop type.
@@ -962,3 +975,103 @@ and browser appearance remain unverified. No new tests or background processes.
 Deploying starts a new learner using existing recordings; enabled-service startup
 removes older policy models and optimizer/scheduler lineages through v4. Current
 lineage is v5. No settings changes or prescribed human driving are required.
+
+### Simplified controls, policy 5
+
+User authorized removal of horn/song/laser/brush motors. Retained 13 seeded shapes:
+four drive zero/nonzero combinations, two camera angle shapes, three headlight
+actions, four raw operating-mode/dock constants. Generic peripherals remain excluded.
+Removed excluded command history/targets via policy allowlist and execution via
+Node allowlist; raw payloads must exactly match one of the four mode commands at
+both boundaries. Controller shutdown now sends only zero-speed drive, leaving
+manual brush/horn state alone. Raw recording and sensor telemetry unchanged.
+Old policies are deleted on enabled startup; resume-v6.pt/schedule-v6.json begin a
+fresh learner using retained recordings. Existing evaluation behavior unchanged.
+Verification passed: production train.py --check-dependencies (policy 5, 13 seeded
+shapes), infer.py --help, changed Node module syntax, focused card ESLint, and diff
+whitespace checks. No live hardware/data available locally; actual learning and
+driving quality remain unverified. No tests added or background processes left.
+
+### Active full learning-path audit (authorized IPI)
+
+User explicitly permits local training/inference with controlled synthetic input;
+there are NO real recordings on this machine. Existing user-control simplification
+edits were present at start and must be preserved. No new schema reset is planned.
+
+Reproduced: production reader with a generated lossless 25 FPS video and 31 valid
+human drive events labeled ZERO commands. Old sampler used 120ms frame intervals
+with only 100ms target coverage. Fixed causal 100ms grid shared by dataset/inference
+uses latest past image; no future pixels. The repaired fixture preserves all 31
+commands. Sensor cutoff is strictly before decision in both paths.
+
+Original objective experiment: 12 easy colored-frame samples, 120 steps, seed 7,
+4 threads: loss 0.8206 -> 0.1996, mean true-command score 0.5106 -> 0.7268, recall
+at 0.7 reached 2/3; at 80 steps recall was still zero. This disproves a wholly
+broken gradient path; it does not diagnose real-data score suppression. Keep
+existing loss/threshold until real-data diagnostics justify changing them.
+
+New learning.py shares the actual objective for ongoing training and disposable
+probe. Per-job before/after evaluation on <=96 unchanged training examples, full
+held-out score distributions, threshold sweep, training-only frequency and no-op
+baselines, decoded pre-server-limit parameter error, gradient norms and sampled
+versus natural frequencies. Bounded <=20s/80-step memorization probe on <=12
+training examples at first job/every 1000 steps; copy only, never published or
+used for generalization gate. Continues current checkpoint/optimizer lineage.
+
+Audit results (synthetic only):
+- Round-tripped eight representative retained command payloads, including signed
+  drive/servo values, zero-speed drive, headlight and all four mode commands.
+- Lossless 25 FPS recording with 31 command events: old sampler labeled 0; new
+  sampler labeled 31/31, including interleaved ignored sensorStream events.
+- Full production train.py decoded five generated recordings, trained/exported
+  checkpoints, evaluated held-out sessions, and resumed optimizer/weights for a
+  second job. 120 -> 240 cumulative steps improved recall at unchanged 0.7 from
+  0 to 0.2553, with precision 1.0 at step 240. Mean true-event score rose 0.5346
+  -> 0.6303. Gate correctly remained failed. A lower 0.5 threshold had precision
+  0.6629/recall 0.8369 and 120 false-positive actions, so threshold was NOT lowered.
+- Disposable diverse probe covered zero-speed drive, moving drive and servo;
+  reached recall 0.75 after 39 steps under its 20s budget. Sampling was corrected
+  to avoid probing only the first event category. Probe uses a copy and higher
+  learning rate for memorization; never adopts weights or changes main optimizer.
+- Exported trained checkpoint reload, forward inference and decoding passed.
+- Retained recordings had extra removed actuator fields, causing false model
+  incompatibility. Normalize saved and current profiles through controlProfile.
+  Direct Node exercise accepts removed laser/peripheral differences and still
+  rejects a changed retained camera configuration.
+
+Diagnostics now report accepted vs represented command counts, request/dispatch
+delay, maximum sensor/image age; sampled vs natural command frequencies; gradient
+norms; same training-subset before/after scores; per-command score distributions;
+threshold sweep and no-op/training-frequency baselines; pre-server-limit decoded
+parameter errors; and bounded diverse memorization probe. Evaluation-step identity
+separates last evaluation from current-job progress. Compact history avoids
+repeating entire command/score tables for every job. Four concise numeric rows
+were added to expanded details; full evidence is in Copy diagnostics.
+
+The original loss and configured 0.7 threshold remain unchanged because experiments
+did not establish that a different objective/threshold fixes real recordings.
+Promotion keeps existing checks and additionally must beat the simple training-
+frequency baseline F1. One-time retry of old reader-specific no-aligned-window
+rejections preserves optimizer state and unrelated errors.
+
+Limits: actual human footage and rover/RTSP transport unavailable. Isolated held-out
+windows cannot measure sequential latch behavior or physical driving; decoded
+parameter errors are explicitly before Node's rover limits/permission/safety
+handling. Main model learned some synthetic cues but has not demonstrated useful
+autonomous driving. Camera command encoding still uses whole-degree predictions.
+The user must not be asked to drive prescribed routes or perform a manual checklist.
+
+Verification: full synthetic worker jobs and checkpoint inference above passed;
+focused Node syntax/card ESLint/diff checks passed before final report additions.
+One experimental seeded run failed because its runpy launcher omitted worker
+import path; corrected launcher succeeded. No production import failure occurred.
+Synthetic inference-worker media exercise passed: eight prediction batches,
+threshold update to 0.3, retained command types and discrete latching exercised.
+This check used a generated checkpoint and a FIFO media adapter preserving
+synthetic PTS (RTSP wall-clock override disabled in the experiment only); it does
+not validate real network capture timing. Trained checkpoint forward/decode was
+verified separately. Temporary media launchers needed fixes for file lifetime
+and command-string escaping; these were experiment issues, not production fixes.
+No repository tests added. Experiments finished and no background worker remains.
+Final focused Node syntax and diff whitespace checks passed. Card lint passed.
+No additional schema/optimizer reset introduced by this audit.

server/src/configuration/migrations.js

@@ -1,6 +1,6 @@
 // Configuration Database Migrations
 // Purpose: Applies ordered, transactional schema changes to the configuration and administration database.
-// Scope: Owns database structure only; configuration-document evolution belongs to the ordered definition and validation.
+// Scope: Upgrades database structure and stored configuration documents before current-schema validation.
 const migrations = [
   {
     version: 1,
@@ -84,6 +84,34 @@ const migrations = [
       `).run();
     },
   },
+  {
+    version: 4,
+    run(db) {
+      const rows = db.prepare('SELECT id, config_json FROM configuration_revisions').all();
+      const update = db.prepare('UPDATE configuration_revisions SET config_json = ? WHERE id = ?');
+      // Upgrade history as well as the active revision so a settings rollback
+      // cannot reintroduce fields rejected by the new Automations schema.
+      for (const row of rows) {
+        const config = JSON.parse(row.config_json);
+        let changed = false;
+        for (const item of config.homeAssistantActivities?.items || []) {
+          const hasAction = Object.hasOwn(item, 'idleAction');
+          const hasValue = Object.hasOwn(item, 'idleValue');
+          if (!hasAction && !hasValue) continue;
+          item.automations ??= {};
+          item.automations.idle ??= {};
+          // Preserve any explicitly saved new-format setting when both formats
+          // exist, including an empty text value used to clear an entity.
+          if (hasAction && !Object.hasOwn(item.automations.idle, 'action')) item.automations.idle.action = item.idleAction;
+          if (hasValue && !Object.hasOwn(item.automations.idle, 'value')) item.automations.idle.value = item.idleValue;
+          delete item.idleAction;
+          delete item.idleValue;
+          changed = true;
+        }
+        if (changed) update.run(JSON.stringify(config), row.id);
+      }
+    },
+  },
 ];
 
 function applySchemaMigrations(db) {

server/src/services/homeAssistantActivitiesService/actions.js

@@ -1,5 +1,4 @@
-// User and idle commands share one dispatch contract, while only the trusted
-// idle entry point can bypass per-item user locks.
+// User and automated commands share validation; only trusted automation bypasses user permissions.
 const { buildEntity, buildCommand } = require('./entityHelpers');
 
 function assertAccess(actor = {}) {
@@ -9,13 +8,14 @@ function assertAccess(actor = {}) {
 }
 
 function createActions({ getConfig, ha, locks }) {
-  async function execute(id, value, actor, idle = false) {
-    if (!idle) assertAccess(actor);
+  async function execute(id, value, actor, automated = false) {
+    if (!automated) assertAccess(actor);
     const config = getConfig();
     if (!config.enabled) throw new Error('Activity Controls are disabled');
     const item = config.items.find((entry) => entry.id === id);
     if (!item) throw new Error('Unknown activity item');
-    if (!idle && locks.isLocked(id) && !['admin', 'lockdown'].includes(actor.role)) throw new Error('This item is locked');
+    if (!automated && locks.isLocked(id) && !['admin', 'lockdown'].includes(actor.role)) throw new Error('This item is locked');
+    if (!automated && item.readOnly && !['admin', 'lockdown'].includes(actor.role)) throw new Error('This item is read-only for users');
     if (!ha.enabled || !ha.isConnected()) throw new Error('Home Assistant is offline');
     const entity = buildEntity(item, ha.getRawEntitySnapshot(id));
     if (!entity.available) throw new Error('This item is unavailable');
@@ -25,30 +25,34 @@ function createActions({ getConfig, ha, locks }) {
     await ha.callHomeAssistantService(entity.domain, command.service, command.data);
   }
 
-  async function runIdleActions() {
+  async function runAutomation(trigger, shouldRun = () => true) {
     const config = getConfig();
-    if (!config.enabled) return { action: 'homeAssistantActivitiesIdle', skipped: true };
+    if (!config.enabled) return { action: trigger, skipped: true };
     const results = [];
     // Catch per item so one offline integration cannot prevent the remaining
-    // configured cleanup actions from running during this idle window.
+    // configured actions from running for this trigger.
     for (const item of config.items) {
-      if (!item.idleAction || item.idleAction === 'unchanged' || item.readOnly) continue;
+      // Stop remaining writes if presence or mode changed while HA was responding.
+      if (!shouldRun()) break;
+      const setting = item.automations?.[trigger];
+      if (!setting?.action || setting.action === 'unchanged') continue;
       try {
         const entity = buildEntity(item, ha.getRawEntitySnapshot(item.id));
         if (entity.type === 'readOnly') continue;
-        if ((item.idleAction === 'press') !== (entity.type === 'button')) throw new Error('Idle action does not match the control type');
+        if ((setting.action === 'press') !== (entity.type === 'button')) throw new Error('Automation action does not match the control type');
         // The admin form omits an empty optional string. Treat that as empty
         // text so clearing a message works; other types still reject it through
         // their normal value validation rather than silently receiving zero.
-        await execute(item.id, item.idleAction === 'press' ? 'press' : (item.idleValue ?? ''), null, true);
+        await execute(item.id, setting.action === 'press' ? 'press' : (setting.value ?? ''), null, true);
         results.push({ id: item.id, ok: true });
       } catch (error) {
         results.push({ id: item.id, ok: false, error: error.message });
       }
     }
-    return { action: 'homeAssistantActivitiesIdle', results };
+    return { action: trigger, results };
   }
-  return { act: (id, value, actor) => execute(id, value, actor), runIdleActions };
+  return { act: (id, value, actor) => execute(id, value, actor), runAutomation,
+    runIdleActions: () => runAutomation('idle') };
 }
 
 module.exports = { createActions, assertAccess };

server/src/services/homeAssistantActivitiesService/activities.test.js

@@ -29,7 +29,7 @@ test('sensor states, live constraints, and unknown button states are preserved',
   assert.equal(buildEntity({ id: 'button.bell' }, raw('unknown')).available, true);
   assert.equal(buildEntity({ id: 'button.bell' }, raw('unavailable')).available, false);
   assert.equal(buildEntity({ id: 'button.bell' }, null).available, false);
-  assert.equal(buildEntity({ id: 'switch.fan', readOnly: true }, raw('on')).type, 'readOnly');
+  assert.equal(buildEntity({ id: 'switch.fan', readOnly: true }, raw('on')).type, 'toggle');
 });
 
 test('domain dispatch handles native entities and helpers without room-light payloads', async () => {
@@ -87,22 +87,22 @@ test('permissions, locks, availability and allowlist are authoritative', async (
 test('idle ignores user locks, preserves no-op items, and isolates failures', async () => {
   const h = harness([
     { id: 'switch.keep' },
-    { id: 'number.bad', idleAction: 'set', idleValue: '200' },
-    { id: 'switch.fan', idleAction: 'set', idleValue: 'off' },
+    { id: 'number.bad', automations: { idle: { action: 'set', value: '200' } } },
+    { id: 'switch.fan', automations: { idle: { action: 'set', value: 'off' } } },
     // An empty optional text box is omitted by the schema-driven admin form.
-    { id: 'input_text.message', idleAction: 'set' },
-    { id: 'input_button.stop', idleAction: 'press' },
-    { id: 'sensor.humidity', idleAction: 'set', idleValue: '0' },
-    { id: 'switch.readonly', readOnly: true, idleAction: 'set', idleValue: 'off' },
+    { id: 'input_text.message', automations: { idle: { action: 'set' } } },
+    { id: 'input_button.stop', automations: { idle: { action: 'press' } } },
+    { id: 'sensor.humidity', automations: { idle: { action: 'set', value: '0' } } },
+    { id: 'switch.readonly', readOnly: true, automations: { idle: { action: 'set', value: 'off' } } },
   ], {
-    'number.bad': raw('0', { min: 0, max: 10 }), 'switch.fan': raw('on'),
+    'switch.readonly': raw('on'), 'number.bad': raw('0', { min: 0, max: 10 }), 'switch.fan': raw('on'),
     'input_text.message': raw('Welcome', { min: 0, max: 30 }), 'input_button.stop': raw('unknown'),
   });
   h.locked.add('switch.fan');
   const result = await h.actions.runIdleActions();
   assert.equal(result.results[0].ok, false);
-  assert.equal(result.results.filter((entry) => entry.ok).length, 3);
-  assert.deepEqual(h.calls.map((call) => call[2]), [{ entity_id: 'switch.fan' }, { entity_id: 'input_text.message', value: '' }, { entity_id: 'input_button.stop' }]);
+  assert.equal(result.results.filter((entry) => entry.ok).length, 4);
+  assert.deepEqual(h.calls.map((call) => call[2]), [{ entity_id: 'switch.fan' }, { entity_id: 'input_text.message', value: '' }, { entity_id: 'input_button.stop' }, { entity_id: 'switch.readonly' }]);
 });
 
 test('an outstanding service response does not block later commands', async () => {

server/src/services/homeAssistantActivitiesService/configuration.js

@@ -2,6 +2,14 @@
 // never acquire unrelated devices simply because they share a connection.
 const { strictObject, string, boolean } = require('../../configuration/schemaHelpers');
 
+// All triggers use the same inputs so the admin form and execution contract agree.
+function automation(title, description) {
+  return strictObject({
+    action: string({ title: 'Action', enum: ['unchanged', 'set', 'press'], default: 'unchanged' }),
+    value: string({ title: 'Value', description: 'For set: on/off, a number, exact selection, or text. Empty text clears it. Ignored for press.', maxLength: 255 }),
+  }, { title, description });
+}
+
 module.exports = {
   key: 'homeAssistantActivities',
   feature: true,
@@ -18,9 +26,12 @@ module.exports = {
         icon: string({ description: 'Font Awesome name, as used by social links. Blank or invalid names use a fallback.', maxLength: 80, examples: ['FaFan'] }),
         // Match social-link colors so admins can customize tiles through the existing config form.
         color: string({ description: 'Optional six-digit hexadecimal tile color, like social links.', examples: ['#3B82F6'], pattern: '^#[0-9a-fA-F]{6}$' }),
-        readOnly: boolean({ title: 'Read only', default: false, description: 'Display the value without allowing user or idle commands.' }),
-        idleAction: string({ title: 'When idle', enum: ['unchanged', 'set', 'press'], default: 'unchanged', description: 'Leave unchanged, set the idle value, or press a button once. Runs independently of user locks.' }),
-        idleValue: string({ title: 'Idle value', description: 'For set: on/off, a number, exact selection, or text (empty text clears it). Checked against live entity limits when idle runs.', examples: ['0'], maxLength: 255 }),
+        readOnly: boolean({ title: 'Read only for users', default: false, description: 'Ordinary users can only view this item. Admins and automations can still change supported controls.' }),
+        automations: strictObject({
+          idle: automation('When idle', 'Runs after the idle timeout with no users or admins online.'),
+          locked: automation('When server locked', 'Runs on entering admin or lockdown mode.'),
+          online: automation('When users online', 'Runs when a user or admin comes online, or the server reopens with one online. Suppressed in admin and lockdown modes. Spectators do not count.'),
+        }, { title: 'Automations', description: 'One-time actions that bypass item locks and read-only permissions.' }),
       }, { description: 'One independently controlled or read-only activity control item.', required: ['id'] }),
     },
   }, { title: 'Activity Controls', description: 'Generic activity controls and idle behavior, separate from room lighting.', required: ['enabled', 'items'] }),

server/src/services/homeAssistantActivitiesService/entityHelpers.js

@@ -9,7 +9,8 @@ const TYPES = {
 function buildEntity(item, raw, locked = false) {
   const attributes = raw?.attributes || {};
   const domain = item.id.split('.')[0];
-  const type = item.readOnly ? 'readOnly' : TYPES[domain] || 'readOnly';
+  // Read-only is a user permission, not an entity capability: admins and automation still need its write type.
+  const type = TYPES[domain] || 'readOnly';
   // A never-pressed button legitimately reports unknown. Its state is a last
   // press timestamp, not availability or a boolean toggle state.
   const available = Boolean(raw && raw.state !== 'unavailable' && (raw.state !== 'unknown' || type === 'button'));
@@ -17,7 +18,7 @@ function buildEntity(item, raw, locked = false) {
   return {
     id: item.id, name: item.name?.trim() || attributes.friendly_name || item.id,
     // Include presentation settings in session state so all clients share the configured color.
-    icon: item.icon || '', color: item.color || '', domain, type, locked, available,
+    icon: item.icon || '', color: item.color || '', domain, type, readOnly: Boolean(item.readOnly), locked, available,
     state: raw?.state ?? 'unknown', unit: attributes.unit_of_measurement || '',
     min: numeric(attributes.min), max: numeric(attributes.max), step: numeric(attributes.step),
     options: Array.isArray(attributes.options) ? attributes.options.filter((option) => typeof option === 'string') : [],

server/src/services/homeAssistantActivitiesService/index.js

@@ -8,7 +8,7 @@ const { buildEntity } = require('./entityHelpers');
 const { createLocks, resolveItem } = require('./locks');
 const { createActions } = require('./actions');
 const { getRole } = require('../roleService');
-const { getMode } = require('../modeManager');
+const { getMode, modeEvents } = require('../modeManager');
 const logger = require('../../globals/logger').child('homeAssistantActivitiesService');
 
 const events = new EventEmitter();
@@ -24,6 +24,35 @@ const actions = createActions({ getConfig: () => config, ha: {
   callHomeAssistantService: ha.callHomeAssistantService,
 }, locks });
 
+let operatorsOnline = false;
+let lastAutomationState = null;
+let automationRevision = 0;
+
+function refreshAutomation() {
+  const mode = getMode();
+  const locked = mode === 'admin' || mode === 'lockdown';
+  // Remember transitions, not entity values: HA updates and extra connections
+  // must not repeatedly press buttons or undo a manual admin adjustment.
+  const state = config.enabled && snapshotReady && ha.isConnected()
+    ? (locked ? mode : operatorsOnline ? 'online' : null) : null;
+  if (state === lastAutomationState) return;
+  lastAutomationState = state;
+  const revision = ++automationRevision;
+  if (!state) return;
+  actions.runAutomation(locked ? 'locked' : 'online', () => revision === automationRevision)
+    .then((result) => {
+      const failures = result.results?.filter((item) => !item.ok) || [];
+      if (failures.length) logger.warn('Activity automation failed', { trigger: result.action, failures });
+    }).catch((error) => logger.warn('Activity automation failed', { error: error.message }));
+}
+
+function setOperatorsOnline(online) {
+  operatorsOnline = online;
+  refreshAutomation();
+}
+
+modeEvents.on('change', refreshAutomation);
+
 function getState() {
   return {
     enabled: config.enabled,
@@ -53,14 +82,17 @@ function setLocked(query, locked) {
 ha.homeAssistantEvents.on('snapshot', () => {
   snapshotReady = true;
   emitUpdate();
+  refreshAutomation();
 });
 ha.homeAssistantEvents.on('status', () => {
   snapshotReady = false;
   emitUpdate();
+  refreshAutomation();
 });
 registerConfigurationHandler('homeAssistantActivities', (next) => {
   config = next;
   emitUpdate();
+  refreshAutomation();
 });
 // Resolve access on the server for every command; clients only supply item/value.
 io.on('connection', (socket) => {
@@ -76,4 +108,7 @@ io.on('connection', (socket) => {
   });
 });
 
-module.exports = { ...actions, getState, setLocked, activityEvents: events };
+module.exports = { ...actions, getState, setLocked, setOperatorsOnline, activityEvents: events,
+  // A returning operator ends idle cleanup even if earlier HA calls are still settling.
+  runIdleActions: () => actions.runAutomation('idle', () => !operatorsOnline),
+};

server/src/services/idleService/index.js

@@ -7,6 +7,7 @@ const { getRole, roleEvents } = require('../roleService');
 const { IDLE_TIMEOUT_MS } = require('./constants');
 const { runtime } = require('./state');
 const { runIdleActions } = require('./actions');
+const activityControls = require('../homeAssistantActivitiesService');
 
 function getActivitySnapshot() {
   let onlineUsers = 0;
@@ -113,6 +114,8 @@ function scheduleIdleTimer() {
 function refreshIdleState() {
   const activity = getActivitySnapshot();
   logger.info('Idle state refresh', activity);
+  // Share the exact idle presence policy; Activity Controls must not count spectators or invent a second role filter.
+  activityControls.setOperatorsOnline(activity.totalActive > 0);
   if (activity.totalActive > 0) {
     clearIdleTimer();
     if (runtime.idleActionsCompleted) {

server/src/services/roverLearningService/capabilities.js

@@ -1,16 +1,18 @@
 // Match actuator configuration, not a rover's display name or changing telemetry.
 const { isDeepStrictEqual } = require('node:util');
-const POLICY_VERSION = 4;
+const POLICY_VERSION = 5;
 
 function controlProfile(meta = {}) {
-  return Object.fromEntries(['cameraServo', 'headlight', 'laser', 'roomba', 'platform']
+  return Object.fromEntries(['cameraServo', 'headlight', 'roomba', 'platform']
     .map((key) => [key, meta?.[key] ?? null]));
 }
 
 function compatible(model, record) {
   if (model?.specification?.version !== POLICY_VERSION || !record) return false;
   const profile = controlProfile(record.meta);
-  return Array.isArray(model.controlProfiles) && model.controlProfiles.some((saved) => isDeepStrictEqual(saved, profile));
+  // Retained recordings may include removed actuator fields. Compare only
+  // controls this policy actually uses, on both sides of the boundary.
+  return Array.isArray(model.controlProfiles) && model.controlProfiles.some((saved) => isDeepStrictEqual(controlProfile(saved), profile));
 }
 
 module.exports = { POLICY_VERSION, controlProfile, compatible };

server/src/services/roverLearningService/diagnostics.js

@@ -7,7 +7,7 @@ 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',
+for (const file of ['workers/policy.py', 'workers/dataset.py', 'workers/train.py', 'workers/infer.py', 'workers/observations.py', 'workers/learning.py',
   'capabilities.js', 'errors.js', 'storage.js', 'runtime.js', 'models.js', 'driving.js', 'recording.js', 'training.js', 'socketGateway.js', 'diagnostics.js']) {
   hash.update(fs.readFileSync(path.join(__dirname, file)));
 }
@@ -41,11 +41,12 @@ function report(state, controller) {
       sessions: state.recording.sessions.map((item) => ({ startedAtUnixMs: item.startedAt,
         status: item.video, droppedEvents: item.droppedEvents, error: redact(item.error), failureDetails: item.failureDetails || null, stderr: redact(item.stderr), errorCategory: errorCategory(item.error) })) },
     training: { status: training.status, steps: training.steps, examples: training.examples,
-      selection: training.selection || null,
+      selection: training.selection || null, evaluatedSteps: training.evaluatedSteps ?? null,
       loss: training.loss, losses: training.losses || null, publication: training.publication || null,
+      trainingAudit: metrics.trainingAudit || 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,
+        evaluationSeconds: metrics.evaluationSeconds ?? null, probeSeconds: metrics.probeSeconds ?? null, beforeAuditSeconds: metrics.beforeAuditSeconds ?? null, checkpointSeconds: metrics.checkpointSeconds ?? null, cpuSeconds: metrics.cpuSeconds ?? null,
         peakRssMiB: metrics.peakRssMiB ?? null },
       history: training.history || [], rejectionReasons: (training.rejectionReasons || []).map(redact), rejectionCategories: (training.rejectionReasons || []).map(errorCategory),
       error: redact(training.error), errorCategory: errorCategory(training.error), nextAttemptAtUnixMs: training.nextAttemptAt },

server/src/services/roverLearningService/driving.js

@@ -9,7 +9,8 @@ const { recordCommandRequest } = require('../commandService/recording');
 const { acquireModel } = require('./models');
 const { compatible } = require('./capabilities');
 
-const OPERATING = new Set(['drive', 'motors', 'servo', 'headlight', 'laser', 'horn', 'song', 'raw']);
+const OPERATING = new Set(['drive', 'servo', 'headlight', 'raw']);
+const MODE_COMMANDS = new Set([128, 131, 132, 143].map((opcode) => Buffer.from([opcode]).toString('base64')));
 const clamp = (value, min, max) => Math.max(min, Math.min(max, Number(value) || 0));
 
 function canControl(socket, roverId) {
@@ -34,14 +35,12 @@ function createDriving({ root, config, onChange, activity }) {
     if (session.status === 'stopping') return;
     if (!canControl(session.socket, session.roverId)) { stop(session.roverId, 'Control permission lost'); return; }
     if (!OPERATING.has(command?.type)) throw new Error('Unsupported model command');
+    if (command.type === 'raw' && !MODE_COMMANDS.has(command.raw)) throw new Error('Unsupported model operating-mode command');
     const record = roverManager.rovers.get(session.roverId);
     if (command.type === 'drive') {
       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) };
-    }
     if (command.type === 'servo') {
       const servo = record.meta?.cameraServo;
       if (!servo?.enabled) { session.submissions.push({ command, result: 'Servo unavailable' }); return; }
@@ -64,8 +63,6 @@ function createDriving({ root, config, onChange, activity }) {
     // A stop must still reach the rover after turn permission is revoked.
     for (const command of [
       { type: 'drive', driveDirect: { left: 0, right: 0 } },
-      { type: 'motors', motorPwm: { main: 0, side: 0, vacuum: 0 } },
-      { type: 'horn', horn: { action: 'stop' } },
     ]) { try { issueCommand(roverId, command); } catch { /* Rover may already be offline. */ } }
     session.worker.kill('SIGTERM');
     const timer = setTimeout(() => session.worker.kill('SIGKILL'), 2000);

server/src/services/roverLearningService/models.js

@@ -142,8 +142,8 @@ function removeObsoleteModels(root) {
       else { await fs.rm(target, { recursive: true, force: true }); retired.delete(target); }
     }
     // These filenames belong to superseded optimizer/scheduler lineages, not
-    // policy versions. The current command policy uses lineage v5.
-    for (const suffix of ['', '-v1', '-v2', '-v3', '-v4']) {
+    // policy versions. The current command policy uses lineage v6.
+    for (const suffix of ['', '-v1', '-v2', '-v3', '-v4', '-v5']) {
       for (const filename of [`resume${suffix}.pt`, `resume${suffix}.tmp`,
         `schedule${suffix}.json`, `schedule${suffix}.json.tmp`]) {
         await fs.rm(path.join(root, 'training', filename), { force: true });

server/src/services/roverLearningService/training.js

@@ -33,10 +33,17 @@ function createTrainer({ root, config, actionThreshold, logger }) {
         }
       }
       try {
-        const saved = JSON.parse(await fs.readFile(path.join(trainingRoot, 'schedule-v5.json'), 'utf8'));
+        const saved = JSON.parse(await fs.readFile(path.join(trainingRoot, 'schedule-v6.json'), 'utf8'));
         ledger = saved.sessions;
         state = { ...saved.progress, status: 'waiting' };
         if (!ledger || typeof ledger !== 'object') throw new Error('Invalid training schedule');
+        // Retry this reader-specific rejection once after the fixed-grid repair.
+        // Keep optimizer progress and unrelated recording failures intact.
+        if (saved.samplingVersion !== 2) {
+          for (const [id, entry] of Object.entries(ledger)) {
+            if (entry.rejected === 'No complete, aligned observation/action windows') delete ledger[id];
+          }
+        }
       } catch (error) { if (error.code !== 'ENOENT') throw error; }
     } catch (error) {
       startupError = true;
@@ -46,8 +53,8 @@ function createTrainer({ root, config, actionThreshold, logger }) {
   }
 
   async function saveLedger() {
-    const filename = path.join(trainingRoot, 'schedule-v5.json');
-    await fs.writeFile(`${filename}.tmp`, JSON.stringify({ sessions: ledger, progress: state }));
+    const filename = path.join(trainingRoot, 'schedule-v6.json');
+    await fs.writeFile(`${filename}.tmp`, JSON.stringify({ samplingVersion: 2, sessions: ledger, progress: state }));
     await fs.rename(`${filename}.tmp`, filename);
   }
 
@@ -159,6 +166,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,
+              evaluatedSteps: result.trained ? result.steps : state.evaluatedSteps,
               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();
@@ -169,7 +177,9 @@ function createTrainer({ root, config, actionThreshold, logger }) {
           logger.error('Training result persistence failed', { error: error.message, failure: errorDetails(error, 'training result persistence') });
         } finally {
           history.push({ at: Date.now(), status: state.status, steps: state.steps, loss: state.loss,
-            losses: state.losses || null, evaluation: state.evaluation || null,
+            losses: state.losses || null, evaluation: state.evaluation ? Object.fromEntries(
+              ['status', 'windows', 'precision', 'recall', 'f1', 'predictedActions', 'positiveActions', 'numericMse', 'missedStops', 'gateReasons']
+                .map((key) => [key, state.evaluation[key]])) : null,
             elapsedSeconds: state.elapsedSeconds || null });
           if (history.length > 20) history.shift();
           leases.clear();

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

@@ -8,6 +8,7 @@ from pathlib import Path
 
 import av
 import numpy as np
+from observations import decision_frames, history_frames
 
 from policy import SLOTS, VALUES, HEIGHT, WIDTH, SENSORS, command_family, is_stop, encode_command, sensor_vector
 
@@ -74,6 +75,7 @@ def prepare(directory, schema, stats, extend):
     candidate = copy.deepcopy(schema)
     history, targets, unsupported_times = [], [], []
     unsupported = {}
+    dispatch_delays = []
     stats.update({'sensorFrames': len(sensors), 'dispatches': len(dispatches),
                   'failedCommands': len(failed), 'sensorPresentFraction': np.mean([row[1][len(SENSORS):] for row in sensors], axis=0).tolist()})
     for data in dispatches:
@@ -103,11 +105,17 @@ def prepare(directory, schema, stats, extend):
             target_time = request['ts']
             if not 0 <= stamp - target_time <= 1000:
                 raise ValueError('Invalid request/dispatch timing')
+            dispatch_delays.append(stamp - target_time)
             targets.append((target_time, *encoded))
         else:
             # Server safety overrides belong in history, never demonstration
             # targets. Exclude windows around them rather than teaching inaction.
             boundaries.append(stamp)
+    stats['requestToDispatchMs'] = {'mean': sum(dispatch_delays) / max(1, len(dispatch_delays)),
+                                    'max': max(dispatch_delays, default=0)}
+    stats['acceptedCommandEvents'] = len(targets)
+    stats['representedCommandEvents'] = 0
+    stats['commandEventCounts'] = [0] * SLOTS
     stats['unsupportedCommands'] = list(unsupported.values())
     stats['unsupportedCommandEvents'] = len(unsupported_times) // 2
     if not targets and not unsupported_times:
@@ -121,7 +129,7 @@ def windows(directory, schema, cancelled=lambda: False, stats=None, extend=True)
     stats = stats if stats is not None else {}
     metadata, sensors, history, targets, boundaries, unsupported_times = prepare(directory, schema, stats, extend)
     stats.update({'usableWindows': 0, 'commandWindows': 0, 'stopWindows': 0, 'noCommandWindows': 0,
-                  'staleSensorWindows': 0, 'boundaryWindows': 0,
+                  'staleSensorWindows': 0, 'boundaryWindows': 0, 'maxSensorAgeMs': 0, 'maxFrameAgeMs': 0,
                   'unsupportedWindows': 0, 'ambiguousWindows': 0, 'videoFrames': 0, 'maxVideoGapMs': 0})
     sensor_times = [row[0] for row in sensors]
     target_times = [row[0] for row in targets]
@@ -131,8 +139,9 @@ def windows(directory, schema, cancelled=lambda: False, stats=None, extend=True)
     command_times = np.full(SLOTS, -np.inf)
     cursor = 0
     recent = collections.deque(maxlen=32)
-    last_sample, last_pts = -math.inf, -math.inf
+    last_pts = -math.inf
     def frames():
+        nonlocal last_pts
         files = metadata['video'].get('files', ['video.mkv'])
         for name in files:
             if Path(name).name != name:
@@ -140,27 +149,24 @@ def windows(directory, schema, cancelled=lambda: False, stats=None, extend=True)
             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 container.decode(stream):
+                    if frame.pts is None:
+                        raise ValueError('Video frame has no timestamp')
+                    stamp = float(frame.pts * frame.time_base) * 1000
+                    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')
+                    yield frame
 
-    for frame in frames():
+    represented = set()
+    for stamp, frame, captured_at in decision_frames(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')
-        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')
         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]
             # Do not let drive predictions simply copy earlier human/model drive commands.
@@ -168,22 +174,19 @@ def windows(directory, schema, cancelled=lambda: False, stats=None, extend=True)
                 command_values[slot] = values
                 command_times[slot] = when
             cursor += 1
-        sensor_index = bisect.bisect_right(sensor_times, stamp) - 1
+        sensor_index = bisect.bisect_left(sensor_times, stamp) - 1
         if sensor_index < 0 or stamp - sensor_times[sensor_index] > 1000:
             stats['staleSensorWindows'] += 1
             recent.clear()
             continue
+        stats['maxSensorAgeMs'] = max(stats['maxSensorAgeMs'], stamp - sensor_times[sensor_index])
+        stats['maxFrameAgeMs'] = max(stats['maxFrameAgeMs'], stamp - captured_at)
         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])
+        chosen = history_frames(recent, stamp)
         if len(chosen) != 4 or stamp + 100 > metadata['endedAt']:
             continue
         if bisect.bisect_right(unsupported_times, stamp + 100) > bisect.bisect_left(unsupported_times, chosen[0][0]):
@@ -201,7 +204,7 @@ def windows(directory, schema, cancelled=lambda: False, stats=None, extend=True)
         for _, slot, vector in targets[start:end]:
             template = schema[slot]['template']
             family = command_family(template)
-            if ((actions[slot] and (template['type'] in ('headlight', 'laser', 'horn', 'song', 'raw') or not np.array_equal(values[slot], vector)))
+            if ((actions[slot] and (template['type'] in ('headlight', 'raw') or not np.array_equal(values[slot], vector)))
                     or (not actions[slot] and family in families)):
                 ambiguous = True
                 break
@@ -210,6 +213,10 @@ def windows(directory, schema, cancelled=lambda: False, stats=None, extend=True)
         if ambiguous:
             stats['ambiguousWindows'] += 1
             continue
+        represented.update(range(start, end))
+        stats['representedCommandEvents'] = len(represented)
+        for slot in np.flatnonzero(actions):
+            stats['commandEventCounts'][slot] += 1
         stats['usableWindows'] += 1
         stats['commandWindows' if actions.any() else 'noCommandWindows'] += 1
         stats['stopWindows'] += int(any(actions[slot] and is_stop(descriptor) for slot, descriptor in enumerate(schema)))

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

@@ -13,6 +13,7 @@ import time
 import av
 import numpy as np
 import torch
+from observations import decision_frames, history_frames
 
 from policy import Policy, VERSION, SLOTS, VALUES, HEIGHT, WIDTH, SENSORS, select_slots, decode_command, encode_command, sensor_vector
 
@@ -54,23 +55,16 @@ def main():
     sensors, sensor_time = {}, 0
     history = collections.deque(maxlen=32)
     latched = set()
-    sampled = 0
     with av.open(args.url, options={'rtsp_transport': 'tcp', 'timeout': '3000000',
                                    'use_wallclock_as_timestamps': '1'}) as video:
         stream = video.streams.video[0]
         stream.codec_context.thread_count = 1
-        for frame in video.decode(stream):
+        for stamp, frame, captured_at in decision_frames(video.decode(stream)):
             if closed.is_set():
                 return
-            if frame.pts is None:
-                continue
-            stamp = float(frame.pts * frame.time_base) * 1000
             now = time.time() * 1000
-            if not math.isfinite(stamp) or abs(now - stamp) > 2000:
+            if abs(now - captured_at) > 2000:
                 raise ValueError('Video time is stale or not aligned to server time')
-            if stamp - sampled < 99:
-                continue
-            sampled = stamp
             with mutex:
                 while observations and observations[0]['ts'] < stamp:
                     event = observations.popleft()
@@ -94,12 +88,7 @@ def main():
                                     np.isfinite(last_commands).astype(np.float32)))
             pixels = frame.to_ndarray(width=WIDTH, height=HEIGHT, format='rgb24').transpose(2, 0, 1).copy()
             history.append((stamp, pixels, state))
-            selected = []
-            for offset in (900, 600, 300, 0):
-                candidates = [entry for entry in history if entry[0] <= stamp - offset + .01]
-                if not candidates or stamp - offset - candidates[-1][0] > 200:
-                    break
-                selected.append(candidates[-1])
+            selected = history_frames(history, stamp)
             if len(selected) != 4:
                 continue
             before = time.perf_counter()
@@ -116,7 +105,7 @@ def main():
             proposals, commands = [], []
             for slot, descriptor in enumerate(schema):
                 command = decode_command(descriptor, predictions[slot])
-                discrete = command['type'] in ('headlight', 'laser', 'horn', 'song', 'raw')
+                discrete = command['type'] in ('headlight', 'raw')
                 reason = ('Below threshold' if slot not in active else
                           'Competing action' if slot not in winners else
                           'Held discrete action' if discrete and slot in latched else 'Proposed')
@@ -125,7 +114,7 @@ def main():
                     commands.append(command)
                 if slot in winners and discrete:
                     latched.add(slot)
-            print(json.dumps({'kind': 'prediction', 'frameAt': stamp, 'sensorAt': sensor_time,
+            print(json.dumps({'kind': 'prediction', 'frameAt': captured_at, 'decisionAt': stamp, 'sensorAt': sensor_time,
                               'latencyMs': (time.perf_counter() - before) * 1000,
                               'sensorPresent': sensor_vector(sensors)[len(SENSORS):].tolist(),
                               'commands': commands, 'proposals': proposals, 'threshold': args.threshold}, allow_nan=False), flush=True)

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

@@ -0,0 +1,181 @@
+"""Training objective and bounded, observational learning diagnostics."""
+import copy
+import time
+
+import numpy as np
+import torch
+from torch.nn import functional as F
+
+from policy import SLOTS, VALUES, select_slots, decode_command, is_stop
+
+
+def loss_setup(samples, schema):
+    numeric_mask = torch.zeros(SLOTS, VALUES)
+    for slot, descriptor in enumerate(schema):
+        numeric_mask[slot, :len(descriptor['fields'])] = 1
+    positives = np.sum([sample[2] for sample in samples], axis=0)
+    opportunities = np.sum([sample[4] for sample in samples], axis=0)
+    weight = torch.from_numpy(np.clip((opportunities - positives) / np.maximum(positives, 1), 1, 20)).float()
+    return numeric_mask, weight
+
+
+def tensors(samples):
+    images, states, targets, values, available = [torch.from_numpy(np.stack(items)) for items in zip(*samples)]
+    return images.float().div_(255), states, targets, values, available
+
+
+def losses(logits, predictions, targets, values, available, numeric_mask, positive_weight):
+    events = F.binary_cross_entropy_with_logits(logits, targets, pos_weight=positive_weight, reduction='none')
+    events = (events * available).sum() / available.sum().clamp_min(1)
+    mask = numeric_mask.unsqueeze(0) * targets.unsqueeze(-1)
+    parameters = ((predictions - values).square() * mask).sum() / mask.sum().clamp_min(1)
+    return events, parameters
+
+
+def distribution(values):
+    values = np.asarray(values, dtype=np.float64)
+    if not values.size:
+        return {'count': 0, 'mean': None, 'p50': None, 'p95': None, 'max': None}
+    return {'count': int(values.size), 'mean': float(values.mean()),
+            'p50': float(np.quantile(values, .5)), 'p95': float(np.quantile(values, .95)), 'max': float(values.max())}
+
+
+def decisions(scores, schema, threshold):
+    selected = np.zeros_like(scores, dtype=bool)
+    for row, vector in enumerate(scores):
+        for slot in select_slots(vector, schema, threshold):
+            selected[row, slot] = True
+    return selected
+
+
+def counts(predicted, actual):
+    matched = int((predicted & actual).sum())
+    proposed, wanted = int(predicted.sum()), int(actual.sum())
+    empty = ~actual.any(axis=1)
+    return {'predictedActions': proposed, 'positiveActions': wanted, 'matchedActions': matched,
+            'precision': matched / max(1, proposed), 'recall': matched / max(1, wanted),
+            'f1': 2 * matched / max(1, proposed + wanted),
+            'falsePositiveActions': proposed - matched,
+            'noCommandWindows': int(empty.sum()), 'unwantedCommandWindows': int((empty & predicted.any(axis=1)).sum())}
+
+
+def summarize(scores, actual, schema, threshold, frequencies):
+    active_scores = scores[:, :len(schema)]
+    labels = actual[:, :len(schema)]
+    thresholds = sorted(set([.1, .3, .5, .7, .9, threshold]))
+    sweep = [{'threshold': value, **counts(decisions(scores, schema, value), actual)} for value in thresholds]
+    frequent = np.broadcast_to(frequencies, scores.shape)
+    baseline = [{'threshold': value, **counts(decisions(frequent, schema, value), actual)} for value in thresholds]
+    return {'scoreOnHumanCommands': distribution(active_scores[labels]),
+            'scoreOnAbsentCommands': distribution(active_scores[~labels]),
+            'maxScoreOnNoCommandWindows': distribution(active_scores[~actual.any(axis=1)].max(axis=1)) if (~actual.any(axis=1)).any() else distribution([]),
+            'thresholds': sweep, 'baselines': {'sendNothing': counts(np.zeros_like(actual), actual),
+                'trainingFrequency': baseline, 'frequencySource': 'Training windows only; no held-out fitting'},
+            'perCommandScores': [{'slot': slot, 'onHumanCommand': distribution(scores[actual[:, slot], slot]),
+                                   'otherwise': distribution(scores[~actual[:, slot], slot])} for slot in range(len(schema))]}
+
+
+def evaluate(model, samples, schema, threshold, frequencies, batch_size, cancelled):
+    """Same score selection as runtime; isolated windows cannot measure latching."""
+    model.eval()
+    score_rows, target_rows = [], []
+    value_error = value_count = 0
+    decoded = {}
+    with torch.inference_mode():
+        for offset in range(0, len(samples), batch_size):
+            if cancelled():
+                break
+            images, states, targets, values, _ = tensors(samples[offset:offset + batch_size])
+            logits, numeric = model(images, states)
+            if not torch.isfinite(logits).all() or not torch.isfinite(numeric).all():
+                raise ValueError('Nonfinite evaluation output')
+            score_rows.append(logits.sigmoid().numpy())
+            target_rows.append(targets.numpy().astype(bool))
+            for row, slot in targets.nonzero().tolist():
+                descriptor = schema[slot]
+                n = len(descriptor['fields'])
+                value_error += (numeric[row, slot, :n] - values[row, slot, :n]).square().sum().item()
+                value_count += n
+                prediction = decode_command(descriptor, numeric[row, slot])
+                expected = decode_command(descriptor, values[row, slot])
+                for field in descriptor['fields']:
+                    a, b = prediction, expected
+                    for key in field['path']:
+                        a, b = a[key], b[key]
+                    key = descriptor['template']['type'] + '.' + '.'.join(map(str, field['path']))
+                    total, count = decoded.get(key, (0, 0))
+                    decoded[key] = (total + abs(a - b), count + 1)
+    if not score_rows:
+        return {'windows': 0, 'complete': False}
+    scores, actual = np.concatenate(score_rows), np.concatenate(target_rows)
+    predicted = decisions(scores, schema, threshold)
+    stop_slots = [i for i, d in enumerate(schema) if is_stop(d)]
+    move_slots = [i for i, d in enumerate(schema) if d['template']['type'] == 'drive' and not is_stop(d)]
+    stop_count = int(actual[:, stop_slots].sum())
+    stop_match = int((actual[:, stop_slots] & predicted[:, stop_slots]).sum())
+    move_count = int(actual[:, move_slots].sum())
+    move_match = int((actual[:, move_slots] & predicted[:, move_slots]).sum())
+    return {'windows': len(scores), 'complete': len(scores) == len(samples),
+            'stopEvents': stop_count, 'missedStops': stop_count - stop_match,
+            'stopRecall': stop_match / max(1, stop_count),
+            'falseStops': int((predicted[:, stop_slots] & ~actual[:, stop_slots]).sum()),
+            'movementEvents': move_count, 'movementRecall': move_match / max(1, move_count),
+            'commands': [{'slot': i, 'command': d['template'], 'actual': int(actual[:, i].sum()),
+                          'predicted': int(predicted[:, i].sum()),
+                          'matched': int((actual[:, i] & predicted[:, i]).sum())} for i, d in enumerate(schema)],
+            'numericMse': value_error / max(1, value_count),
+            'decodedParameterMaeBeforeServerLimits': {key: total / count for key, (total, count) in decoded.items()},
+            **counts(decisions(scores, schema, threshold), actual),
+            **summarize(scores, actual, schema, threshold, frequencies)}
+
+
+def probe(model, samples, schema, threshold, frequencies, cancelled, max_seconds=20):
+    """Try fitting a small real training subset on a disposable copy, never publish it."""
+    started = time.monotonic()
+    # Include both event and quiet windows; source is training only. No random
+    # state/optimizer changes escape into the continuing learner.
+    quiet = [row for row in samples if not row[2].any()][:4]
+    # A reservoir is grouped by category: taking its first events could probe
+    # only zero-speed drive and falsely suggest all controls were learnable.
+    by_slot = [[i for i, row in enumerate(samples) if row[2][slot]] for slot in range(len(schema))]
+    chosen = []
+    for rank in range(8):
+        for indices in by_slot:
+            if rank < len(indices) and indices[rank] not in chosen and len(chosen) < 8:
+                chosen.append(indices[rank])
+    subset = [samples[i] for i in chosen] + quiet
+    if not chosen:
+        return {'status': 'No command examples', 'steps': 0}
+    with torch.random.fork_rng(devices=[]):
+        candidate = copy.deepcopy(model)
+        optimizer = torch.optim.AdamW(candidate.parameters(), lr=.001)
+        numeric_mask, weights = loss_setup(subset, schema)
+        images, states, targets, values, available = tensors(subset)
+        initial_loss = final_loss = gradient_norm = None
+        completed = 0
+        expired = lambda: cancelled() or time.monotonic() - started >= max_seconds
+        for _ in range(80):
+            if expired():
+                break
+            candidate.train()
+            optimizer.zero_grad(set_to_none=True)
+            logits, predictions = candidate(images, states)
+            event_loss, value_loss = losses(logits, predictions, targets, values, available, numeric_mask, weights)
+            loss = event_loss + value_loss
+            if not torch.isfinite(loss):
+                raise ValueError('Nonfinite learning probe loss')
+            if initial_loss is None:
+                initial_loss = loss.item()
+            loss.backward()
+            norm = torch.nn.utils.clip_grad_norm_(candidate.parameters(), 1)
+            if not torch.isfinite(norm):
+                raise ValueError('Nonfinite learning probe gradient')
+            gradient_norm = float(norm)
+            optimizer.step()
+            final_loss = loss.item()
+            completed += 1
+        result = evaluate(candidate, subset, schema, threshold, frequencies, 16, cancelled)
+    return {'status': 'completed' if completed == 80 else 'budget limited', 'steps': completed,
+            'windows': len(subset), 'initialLoss': initial_loss, 'lastStepLoss': final_loss,
+            'lastGradientNormBeforeClipping': gradient_norm, 'elapsedSeconds': time.monotonic() - started,
+            'evaluation': result, 'meaning': 'Memorization on a disposable copy of training data, not generalization'}

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

@@ -0,0 +1,45 @@
+"""Shared causal 10 Hz sampling for recorded and live video."""
+import math
+
+
+def decision_frames(frames):
+    """Yield decision time, latest past image, and its actual capture timestamp.
+
+    Sampling every third/fourth decoded frame leaves holes at rates such as 25
+    FPS. A fixed grid covers each 100ms command interval exactly once. A frame
+    after the decision can release the pending tick, but is never its input.
+    """
+    previous = None
+    previous_time = None
+    tick = None
+    for frame in frames:
+        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 (previous_time is not None and stamp < previous_time):
+            raise ValueError('Invalid/nonmonotonic video timestamps')
+        if tick is None:
+            tick = stamp
+        while tick < stamp - .01:
+            # Do not synthesize observations over missing/stalled video. History
+            # selection will reject a temporal gap until fresh history exists.
+            if previous is not None and tick - previous_time <= 200:
+                yield tick, previous, previous_time
+            elif stamp - tick > 200:
+                tick += max(1, math.floor((stamp - tick) / 100)) * 100
+                continue
+            tick += 100
+        if abs(stamp - tick) <= .01:
+            yield tick, frame, stamp
+            tick += 100
+        previous, previous_time = frame, stamp
+
+
+def history_frames(history, stamp):
+    selected = []
+    for offset in (900, 600, 300, 0):
+        candidates = [row for row in history if row[0] <= stamp - offset + .01]
+        if not candidates or stamp - offset - candidates[-1][0] > 200:
+            return []
+        selected.append(candidates[-1])
+    return selected

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

@@ -14,12 +14,13 @@ import numpy as np
 import torch
 from torch import nn
 
-VERSION = 4
+VERSION = 5
 SLOTS = 64
 VALUES = 8
 FRAMES = 4
 HEIGHT, WIDTH = 120, 160
-OPERATING_TYPES = {"drive", "motors", "servo", "headlight", "laser", "horn", "song", "raw"}
+OPERATING_TYPES = {"drive", "servo", "headlight", "raw"}
+MODE_COMMANDS = tuple(base64.b64encode(bytes([opcode])).decode() for opcode in (128, 131, 132, 143))
 SENSORS = [
     ("bumpsAndWheelDrops.bumpLeft", 1), ("bumpsAndWheelDrops.bumpRight", 1),
     ("bumpsAndWheelDrops.wheelDropLeft", 1), ("bumpsAndWheelDrops.wheelDropRight", 1),
@@ -56,29 +57,28 @@ def sensor_vector(sensors):
 def describe_command(command):
     if command.get('type') not in OPERATING_TYPES:
         return None
+    # Raw is restricted to the existing mode/dock buttons, not arbitrary OI
+    # packets that could reintroduce excluded actuators or songs.
+    if command['type'] == 'raw' and command.get('raw') not in MODE_COMMANDS:
+        return None
     if len(json.dumps(command)) > 4096:
         raise ValueError('Operating command exceeds 4096-byte schema limit')
     values, fields = [], []
 
     def visit(value, path=()):
-        if isinstance(value, list) and command['type'] == 'horn' and path == ('horn', 'freqs'):
-            return [visit(item, (*path, index)) for index, item in enumerate(value)]
         if isinstance(value, dict):
             return {key: visit(item, (*path, key)) for key, item in sorted(value.items())}
-        # Exact actuator zeros preserve demonstrated releases. Horn frequency
-        # zeros are parameters so changing a chord does not require a new slot.
-        if value == 0 and command['type'] != 'horn':
+        # Exact actuator zeros preserve demonstrated releases.
+        if value == 0:
             return copy.deepcopy(value)
         if isinstance(value, (int, float)) and not isinstance(value, bool):
             if not math.isfinite(value) or len(values) >= VALUES:
                 raise ValueError('Nonfinite command or too many numeric control fields')
-            scale = {'drive': 500, 'motors': 255, 'servo': 180,
-                     'horn': 1000, 'song': 127}.get(command['type'], 1)
-            fields.append({'path': list(path), 'scale': scale, 'integer': command['type'] in ('drive', 'motors', 'servo', 'horn') or isinstance(value, int)})
+            scale = {'drive': 500, 'servo': 180}.get(command['type'], 1)
+            fields.append({'path': list(path), 'scale': scale, 'integer': command['type'] in ('drive', 'servo') or isinstance(value, int)})
             values.append(normalize(value, scale))
             return {'$number': len(values) - 1}
-        # Arrays (e.g. songs) are discrete exact demonstrations, not variable
-        # length regression outputs. Never reinterpret raw OI bytes as numbers.
+        # Mode bytes and named actions remain exact constants.
         return copy.deepcopy(value)
 
     template = visit(command)
@@ -105,24 +105,17 @@ def encode_command(command, schema, extend=False):
 
 
 def initial_schema():
-    # Match commandPipeline.js envelopes, OI_COMMANDS and the four-frequency
-    # horn control. Values are examples describing fields, not training labels.
+    # Match the retained commandPipeline.js controls. Values describe fields,
+    # not artificial demonstrations; zero-speed drive remains a drive command.
     commands = []
     for left, right in itertools.product((0, 1), repeat=2):
         commands.append({'type': 'drive', 'driveDirect': {'left': left, 'right': right}})
-    for main, side, vacuum in itertools.product((0, 1), repeat=3):
-        commands.append({'type': 'motors', 'motorPwm': {'main': main, 'side': side, 'vacuum': vacuum}})
     for angle in (0, 1):
         commands.append({'type': 'servo', 'servo': {'angle': angle}})
-    for kind in ('headlight', 'laser'):
-        for action in ('on', 'off', 'toggle'):
-            commands.append({'type': kind, kind: {'action': action}})
-    for waveform in ('sine', 'saw'):
-        commands.append({'type': 'horn', 'horn': {'action': 'start', 'waveform': waveform, 'freqs': [1, 1, 1, 1]}})
-    commands.append({'type': 'horn', 'horn': {'action': 'stop'}})
-    for opcode in (128, 131, 132, 143):
-        commands.append({'type': 'raw', 'raw': base64.b64encode(bytes([opcode])).decode()})
-    commands.append({'type': 'song', 'song': {'notes': [{'note': 60, 'duration': 8}]}})
+    for action in ('on', 'off', 'toggle'):
+        commands.append({'type': 'headlight', 'headlight': {'action': action}})
+    for raw in MODE_COMMANDS:
+        commands.append({'type': 'raw', 'raw': raw})
     schema = []
     for command in commands:
         encode_command(command, schema, extend=True)
@@ -188,7 +181,7 @@ class Policy(nn.Module):
 
 
 def specification(schema):
-    return {'version': VERSION, 'architecture': 'spatial-cnn-gru-events-v4',
+    return {'version': VERSION, 'architecture': 'spatial-cnn-gru-events-v5',
             'frames': FRAMES, 'height': HEIGHT, 'width': WIDTH,
             'historyStrideMs': 300, 'decisionIntervalMs': 100,
             'stateSize': STATE_SIZE, 'sensorFields': SENSORS,

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

@@ -39,7 +39,7 @@ def main():
     import numpy as np
     import av
     import torch
-    from torch.nn import functional as functional
+    from learning import loss_setup, tensors, losses as policy_losses, evaluate, probe
     from dataset import windows
     from policy import Policy, VERSION, SLOTS, VALUES, specification, is_stop, select_slots, initial_schema
 
@@ -72,12 +72,13 @@ def main():
     optimizer = torch.optim.AdamW(model.parameters(), lr=0.0001)
     # The command-event policy has different inputs, targets and weights. Keep its
     # optimizer lineage separate; reuse recordings with their unchanged partition.
-    resume_path = root / 'training' / 'resume-v5.pt'
+    resume_path = root / 'training' / 'resume-v6.pt'
     schema, steps, examples, last_published = initial_schema(), 0, 0, 0
     trained_rovers = set()
     control_profiles = []
     distinct_windows = {}
     published_windows = 0
+    last_probe_steps, learning_probe = 0, None
     if resume_path.exists():
         saved = torch.load(resume_path, map_location='cpu', weights_only=True)
         if saved['version'] != VERSION:
@@ -91,6 +92,8 @@ def main():
         control_profiles = saved.get('controlProfiles', [])
         distinct_windows = saved.get('distinctWindows', {})
         published_windows = saved.get('publishedWindows', 0)
+        last_probe_steps = saved.get('lastProbeSteps', 0)
+        learning_probe = saved.get('learningProbe')
     cpu_started = time.process_time()
     decode_started = time.monotonic()
     dataset_stats = []
@@ -171,15 +174,16 @@ def main():
         return
 
     decode_seconds = time.monotonic() - decode_started
+    audit_started = time.monotonic()
+    numeric_mask, positive_weight = loss_setup(pool, schema)
+    training_rows = [row for row in dataset_stats if row.get('partition') == 'training']
+    natural_count = sum(row.get('usableWindows', 0) for row in training_rows)
+    frequencies = np.sum([row.get('commandEventCounts', [0] * SLOTS) for row in training_rows], axis=0) / max(1, natural_count)
+    audit_subset = pool[::max(1, len(pool) // 96)][:96]
+    before_training = evaluate(model, audit_subset, schema, job['actionThreshold'], frequencies, config['batchSize'], cancelled)
+    before_audit_seconds = time.monotonic() - audit_started
     train_started = time.monotonic()
-    numeric_mask = torch.zeros(SLOTS, VALUES)
-    for slot, descriptor in enumerate(schema):
-        numeric_mask[slot, :len(descriptor['fields'])] = 1
-    # Balance sparse events without assuming that every time step should fire
-    # every control. This is training loss, not a driving-quality evaluation.
-    positives = np.sum([sample[2] for sample in pool], axis=0)
-    opportunities = np.sum([sample[4] for sample in pool], axis=0)
-    positive_weight = torch.from_numpy(np.clip((opportunities - positives) / np.maximum(positives, 1), 1, 20)).float()
+    gradient_norms = []
     losses = []
     model.train()
     for _ in range(config['passesPerJob']):
@@ -188,20 +192,18 @@ 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 = images.float().div_(255)
+            images, states, targets, values, available = tensors(batch)
             optimizer.zero_grad(set_to_none=True)
             action_logits, predictions = model(images, states)
-            event_loss = functional.binary_cross_entropy_with_logits(action_logits, targets,
-                                                                       pos_weight=positive_weight, reduction='none')
-            event_loss = (event_loss * available).sum() / available.sum().clamp_min(1)
-            mask = numeric_mask.unsqueeze(0) * targets.unsqueeze(-1)
-            value_loss = ((predictions - values).square() * mask).sum() / mask.sum().clamp_min(1)
+            event_loss, value_loss = policy_losses(action_logits, predictions, targets, values, available, numeric_mask, positive_weight)
             loss = event_loss + value_loss
             if not torch.isfinite(loss):
                 raise ValueError('Nonfinite training loss; checkpoint not saved')
             loss.backward()
-            torch.nn.utils.clip_grad_norm_(model.parameters(), 1)
+            gradient_norm = torch.nn.utils.clip_grad_norm_(model.parameters(), 1)
+            if not torch.isfinite(gradient_norm):
+                raise ValueError('Nonfinite training gradient; checkpoint not saved')
+            gradient_norms.append(float(gradient_norm))
             optimizer.step()
             steps += 1
             examples += len(batch)
@@ -226,58 +228,16 @@ def main():
                   'windows': len(validation), 'unsupportedCommandEvents': unsupported_events,
                   'unsupportedWindows': unsupported_windows}
     if validation and not cancelled():
-        model.eval()
-        true_positive = predicted_positive = actual_positive = 0
-        value_error = value_count = evaluated = 0
-        stop_actual = stop_detected = false_stops = 0
-        drive_actual = drive_detected = 0
-        empty_windows = unwanted_windows = 0
-        stop_slots = [slot for slot, descriptor in enumerate(schema) if is_stop(descriptor)]
-        drive_slots = [slot for slot, descriptor in enumerate(schema)
-                       if descriptor['template']['type'] == 'drive' and not is_stop(descriptor)]
-        per_command = [{'slot': slot, 'command': descriptor['template'], 'actual': 0,
-                        'predicted': 0, 'matched': 0} for slot, descriptor in enumerate(schema)]
-        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 = torch.zeros_like(targets, dtype=torch.bool)
-                # Use the same threshold and competing-command selection as live
-                # inference. Reservoir samples cannot assess temporal latching or
-                # closed-loop motion; report this explicitly rather than implying it.
-                for row, scores in enumerate(logits.sigmoid().tolist()):
-                    for slot in select_slots(scores, schema, job['actionThreshold']):
-                        predicted[row, slot] = True
-                actual = targets.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()
-                empty = ~actual.any(dim=1)
-                empty_windows += empty.sum().item()
-                unwanted_windows += (empty & predicted.any(dim=1)).sum().item()
-                if stop_slots:
-                    stop_actual += actual[:, stop_slots].sum().item()
-                    stop_detected += (actual[:, stop_slots] & predicted[:, stop_slots]).sum().item()
-                    false_stops += (predicted[:, stop_slots] & ~actual[:, stop_slots]).sum().item()
-                if drive_slots:
-                    drive_actual += actual[:, drive_slots].sum().item()
-                    drive_detected += (actual[:, drive_slots] & predicted[:, drive_slots]).sum().item()
-                for item in per_command:
-                    slot = item['slot']
-                    item['actual'] += actual[:, slot].sum().item()
-                    item['predicted'] += predicted[:, slot].sum().item()
-                    item['matched'] += (actual[:, slot] & predicted[:, slot]).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)
-        stop_recall = stop_detected / max(1, stop_actual)
+        measured = evaluate(model, validation, schema, job['actionThreshold'], frequencies, config['batchSize'], cancelled)
+        evaluated = measured['windows']
+        precision, recall = measured.get('precision', 0), measured.get('recall', 0)
+        actual_positive = measured.get('positiveActions', 0)
+        mse = measured.get('numericMse', 0)
+        stop_actual = measured.get('stopEvents', 0)
+        stop_recall = measured.get('stopRecall', 0)
+        drive_actual = measured.get('movementEvents', 0)
+        drive_recall = measured.get('movementRecall', 0)
+        frequency_f1 = max((row['f1'] for row in measured.get('baselines', {}).get('trainingFrequency', [])), default=0)
         held_out_sessions = sum(row.get('partition') == 'heldOut' and not row.get('skipReason') for row in dataset_stats)
         gate_reasons = []
         for condition, reason in [
@@ -288,27 +248,38 @@ def main():
             (actual_positive >= 32, 'Need 32 held-out command events'),
             (stop_actual >= 8, 'Need eight held-out stop events'),
             (drive_actual >= 16, 'Need 16 held-out movement commands'),
-            (drive_detected / max(1, drive_actual) >= .5, 'Movement command recall below 0.5'),
+            (drive_recall >= .5, 'Movement command recall below 0.5'),
             (precision >= .5, 'Command precision below 0.5'),
             (recall >= .5, 'Command recall below 0.5'),
             (stop_recall >= .5, 'Stop recall below 0.5'),
-            (value_count == 0 or mse <= .1, 'Command parameter error above 0.1'),
+            (measured.get('f1', 0) > frequency_f1, 'Command F1 does not beat the training-frequency baseline'),
+            (mse <= .1, 'Command parameter error above 0.1'),
         ]:
             if not condition:
                 gate_reasons.append(reason)
-        evaluation = {'status': 'imitation gate not met' if gate_reasons else 'passed imitation gate',
+        evaluation = {**measured, 'status': 'imitation gate not met' if gate_reasons else 'passed imitation gate',
                       'unsupportedCommandEvents': unsupported_events, 'unsupportedWindows': unsupported_windows,
-                      'method': 'Held-out 100ms command events; not a closed-loop rollout',
+                      'method': 'Isolated held-out 100ms events; live selection, no temporal latch/physical rollout',
                       'threshold': job['actionThreshold'], 'heldOutSessions': held_out_sessions,
-                      'windows': evaluated, 'precision': precision, 'recall': recall,
-                      'numericMse': mse, 'positiveActions': actual_positive,
-                      'predictedActions': predicted_positive, 'falsePositiveActions': predicted_positive - true_positive,
-                      'stopEvents': stop_actual, 'missedStops': stop_actual - stop_detected,
-                      'stopRecall': stop_recall, 'falseStops': false_stops,
-                      'movementEvents': drive_actual, 'movementRecall': drive_detected / max(1, drive_actual),
-                      'noCommandWindows': empty_windows, 'unwantedCommandWindows': unwanted_windows,
-                      'commands': per_command, 'gateReasons': gate_reasons}
-    evaluation_seconds = time.monotonic() - evaluation_started
+                      'gateReasons': gate_reasons}
+    after_training = evaluate(model, audit_subset, schema, job['actionThreshold'], frequencies, config['batchSize'], cancelled)
+    probe_seconds = 0
+    if not cancelled() and (learning_probe is None or steps - last_probe_steps >= 1000):
+        remaining = config['maxJobSeconds'] - (time.monotonic() - started)
+        if remaining > 10:
+            probe_started = time.monotonic()
+            learning_probe = probe(model, audit_subset, schema, job['actionThreshold'], frequencies,
+                                   cancelled, max_seconds=min(20, remaining / 2))
+            learning_probe['atTrainingStep'] = steps
+            last_probe_steps = steps
+            probe_seconds = time.monotonic() - probe_started
+    training_audit = {'before': before_training, 'after': after_training,
+                      'gradientNormBeforeClipping': {'min': min(gradient_norms), 'max': max(gradient_norms)},
+                      'positiveWeights': positive_weight[:len(schema)].tolist(),
+                      'naturalTrainingCommandFrequency': frequencies[:len(schema)].tolist(),
+                      'sampledCommandFrequency': np.mean([row[2] for row in pool], axis=0)[:len(schema)].tolist(),
+                      'learningProbe': learning_probe}
+    evaluation_seconds = time.monotonic() - evaluation_started - probe_seconds
     publication = {'distinctMinutes': unique_count / 600,
                    'newMinutes': (unique_count - published_windows) / 600,
                    'minimumMinutes': config['minimumDrivingMinutes'],
@@ -361,9 +332,9 @@ def main():
         'schema': schema, 'steps': steps, 'examples': examples,
         'lastPublished': last_published, 'trainedRovers': sorted(trained_rovers),
         'controlProfiles': control_profiles, 'distinctWindows': distinct_windows,
-        'publishedWindows': published_windows,
+        'publishedWindows': published_windows, 'lastProbeSteps': last_probe_steps, 'learningProbe': learning_probe,
     }, resume_path)
-    diagnostics = {'dataset': dataset_stats, 'decodeSeconds': decode_seconds,
+    diagnostics = {'dataset': dataset_stats, 'trainingAudit': training_audit, 'probeSeconds': probe_seconds, 'beforeAuditSeconds': before_audit_seconds, '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,

webui/src/components/HomeAssistantActivitiesPanel/index.jsx

@@ -24,8 +24,9 @@ function ActivityTile({ entity, connected, allowed, admin }) {
   // Translucent fills preserve text contrast; an amber border still identifies
   // locked tiles regardless of the admin's chosen color.
   const color = /^#[0-9a-f]{6}$/i.test(entity.color) ? entity.color : '#3b82f6';
-  const Control = controls[entity.type] || ReadOnlyControl;
-  const disabled = !connected || !entity.available || !allowed || (entity.locked && !admin);
+  // Ordinary users see only the value; admins retain the entity's actual input.
+  const Control = entity.readOnly && !admin ? ReadOnlyControl : controls[entity.type] || ReadOnlyControl;
+  const disabled = !connected || !entity.available || !allowed || ((entity.locked || entity.readOnly) && !admin);
   return <div className="flex min-w-0 flex-col gap-0.5 rounded-sm border px-0.5 py-0.5"
     style={{ borderColor: entity.locked ? '#b45309' : `${color}aa`, backgroundColor: `${color}33` }}>
     <div className="-mx-0.5 -mt-0.5 flex min-h-5 min-w-0 items-center gap-0.5 rounded-t px-0.5 py-0.5 text-white" style={{ backgroundColor: `${color}66` }}>

webui/src/components/vip/VipRoverLearningCard.jsx

@@ -133,7 +133,7 @@ function LearningCard({ roverId }) {
       <div className="flex flex-wrap items-center justify-center gap-1">
         <select aria-label="Rover model" className={fieldClass} value={chosen?.id || ''} onChange={(event) => setSelectedId(event.target.value)} disabled={working || !models.length}>
           {!models.length && <option value="">No published models yet</option>}
-          {models.map((model) => <option key={model.id} value={model.id}>{favorites.includes(model.id) ? '★ ' : ''}{model.name}{model.experimental ? ' (experimental)' : ''}{model.compatible ? '' : model.formatVersion !== 4 ? ' (older model format)' : ' (incompatible)'}</option>)}
+          {models.map((model) => <option key={model.id} value={model.id}>{favorites.includes(model.id) ? '★ ' : ''}{model.name}{model.experimental ? ' (experimental)' : ''}{model.compatible ? '' : model.formatVersion !== 5 ? ' (older model format)' : ' (incompatible)'}</option>)}
         </select>
         <button type="button" className="button-dark text-sm disabled:opacity-50" aria-pressed={Boolean(chosen && favorites.includes(chosen.id))} disabled={!chosen} onClick={favorite}>
           {chosen && favorites.includes(chosen.id) ? '★ Favorite' : '☆ Favorite'}
@@ -220,6 +220,10 @@ function LearningCard({ roverId }) {
             <Row label="Latest update status">{training?.publication?.reason || 'Collecting data'}</Row>
             <Row label="Held-out evaluation">{training?.evaluation?.status || 'Waiting for held-out data'}</Row>
             <Row label="Held-out precision / recall">{training?.evaluation?.precision == null ? 'Unavailable' : `${training.evaluation.precision.toFixed(3)} / ${training.evaluation.recall.toFixed(3)}`}</Row>
+            <Row label="Evaluated training step">{display(training?.evaluatedSteps)}</Row>
+            <Row label="Mean / peak score on human commands">{training?.evaluation?.scoreOnHumanCommands?.mean == null ? 'Unavailable' : `${training.evaluation.scoreOnHumanCommands.mean.toFixed(3)} / ${training.evaluation.scoreOnHumanCommands.max.toFixed(3)}`}</Row>
+            <Row label="Training sample recall before / after">{training?.diagnostics?.trainingAudit?.after?.recall == null ? 'Unavailable' : `${training.diagnostics.trainingAudit.before.recall?.toFixed(3) ?? 'Unavailable'} / ${training.diagnostics.trainingAudit.after.recall.toFixed(3)}`}</Row>
+            <Row label="Memorization probe recall / steps">{training?.diagnostics?.trainingAudit?.learningProbe?.evaluation?.recall == null ? 'Unavailable' : `${training.diagnostics.trainingAudit.learningProbe.evaluation.recall.toFixed(3)} / ${training.diagnostics.trainingAudit.learningProbe.steps}`}</Row>
             <Row label="Steps between Latest updates">{display(limits?.training.checkpointEverySteps)}</Row>
           </dl>{training?.error && <p className="break-words text-amber-300">{training.error}</p>}
             {training?.reason && <p>{training.reason}</p>}