From 5a0017cd9a09004d3a16763d36b6c9a0f30f82ab Mon Sep 17 00:00:00 2001
From: Cam Proudlock <cam.proudlock@gmail.com>
Date: Fri, 7 Aug 2026 14:30:18 -0400
Subject: [PATCH] backups plugin with NTLARS revision history; shopfloor
 dashboard upcoming cards and state chips

---
 frontend/src/views/ShopfloorDashboard.vue     | 148 +++-
 plugins/backups/__init__.py                   |   5 +
 plugins/backups/api/__init__.py               |   5 +
 plugins/backups/api/routes.py                 | 306 +++++++++
 plugins/backups/frontend/routes.js            |  21 +
 .../backups/frontend/views/BackupHistory.vue  | 210 ++++++
 plugins/backups/manifest.json                 |  14 +
 plugins/backups/migrations/env.py             |  16 +
 plugins/backups/migrations/script.py.mako     |  24 +
 .../versions/0001_backups_baseline.py         |  61 ++
 plugins/backups/models/__init__.py            |   5 +
 plugins/backups/models/backup.py              | 123 ++++
 plugins/backups/plugin.py                     | 332 +++++++++
 plugins/backups/services/__init__.py          |   6 +
 plugins/backups/services/dncinfo.py           | 161 +++++
 plugins/backups/services/ntlars.py            | 308 +++++++++
 plugins/backups/services/registry.py          | 237 +++++++
 plugins/backups/services/retention.py         |  68 ++
 scripts/import_ntlars_backups.py              | 137 ++++
 shopdb/plugins/alembic_template.py            |   1 +
 tests/test_plugin_migrations.py               |   2 +
 tests/test_plugins/test_backups.py            | 633 ++++++++++++++++++
 22 files changed, 2818 insertions(+), 5 deletions(-)
 create mode 100644 plugins/backups/__init__.py
 create mode 100644 plugins/backups/api/__init__.py
 create mode 100644 plugins/backups/api/routes.py
 create mode 100644 plugins/backups/frontend/routes.js
 create mode 100644 plugins/backups/frontend/views/BackupHistory.vue
 create mode 100644 plugins/backups/manifest.json
 create mode 100644 plugins/backups/migrations/env.py
 create mode 100644 plugins/backups/migrations/script.py.mako
 create mode 100644 plugins/backups/migrations/versions/0001_backups_baseline.py
 create mode 100644 plugins/backups/models/__init__.py
 create mode 100644 plugins/backups/models/backup.py
 create mode 100644 plugins/backups/plugin.py
 create mode 100644 plugins/backups/services/__init__.py
 create mode 100644 plugins/backups/services/dncinfo.py
 create mode 100644 plugins/backups/services/ntlars.py
 create mode 100644 plugins/backups/services/registry.py
 create mode 100644 plugins/backups/services/retention.py
 create mode 100644 scripts/import_ntlars_backups.py
 create mode 100644 tests/test_plugins/test_backups.py

diff --git a/frontend/src/views/ShopfloorDashboard.vue b/frontend/src/views/ShopfloorDashboard.vue
index 2527e33c..d182cb92 100644
--- a/frontend/src/views/ShopfloorDashboard.vue
+++ b/frontend/src/views/ShopfloorDashboard.vue
@@ -37,8 +37,10 @@
             v-for="n in group.items"
             :key="n.notificationid"
             class="banner-strip"
+            :class="{ pending: n.upcoming || n.resolved }"
             :style="{ backgroundColor: getTypeColor(n.typecolor) }"
           >
+            <span v-if="stateLabel(n)" class="state-chip">{{ stateLabel(n) }}</span>
             {{ n.notification }}
           </div>
         </section>
@@ -47,14 +49,15 @@
         <section v-else-if="group.displaystyle === 'carousel'" class="recognition-section">
           <div
             class="section-title recognition"
-            :style="{ backgroundColor: getTypeColor(group.typecolor), color: textOn(group.typecolor) }"
+            :class="{ mixed: isMixed(group) }"
+            :style="headingStyle(group)"
           >{{ group.typename }}</div>
           <div class="recognition-carousel">
             <div
               v-for="(rec, idx) in group.items"
               :key="`${rec.notificationid}-${rec.employeesso}`"
               class="recognition-card"
-              :class="{ active: idx === carouselIndex(group) }"
+              :class="{ active: idx === carouselIndex(group), pending: rec.upcoming || rec.resolved }"
               :style="{ '--accent': getTypeColor(rec.typecolor) }"
             >
               <!-- A type that names no employee is a message, not a person:
@@ -78,17 +81,41 @@
                 <div v-if="hasEmployee(rec)" class="recognition-header">
                   <div class="recognition-name">{{ rec.employeename }}</div>
                 </div>
+                <!-- On a shared category row the heading names the category, so
+                     the card has to say which type it is. Named, not just
+                     coloured: colour alone is no use to a colourblind reader
+                     across a 1920 board. -->
+                <div class="chip-row">
+                  <span
+                    v-if="isMixed(group)"
+                    class="type-chip"
+                    :style="{ backgroundColor: getTypeColor(rec.typecolor), color: textOn(rec.typecolor) }"
+                  >{{ rec.typename }}</span>
+                  <span v-if="stateLabel(rec)" class="state-chip">{{ stateLabel(rec) }}</span>
+                </div>
                 <div class="recognition-message">{{ rec.notification }}</div>
               </div>
             </div>
           </div>
+          <!-- One pip per card, so a reader knows how many are in the rotation
+               and how long until theirs comes back round. -->
+          <div v-if="group.items.length > 1" class="carousel-pips">
+            <span
+              v-for="(rec, idx) in group.items"
+              :key="`pip-${rec.notificationid}-${rec.employeesso}`"
+              class="pip"
+              :class="{ on: idx === carouselIndex(group) }"
+              :style="{ backgroundColor: getTypeColor(rec.typecolor) }"
+            ></span>
+          </div>
         </section>
 
         <!-- Grid - a row of tiles, cycling a page at a time -->
         <section v-else class="recert-section">
           <div
             class="section-title recert-title"
-            :style="{ backgroundColor: getTypeColor(group.typecolor), color: textOn(group.typecolor) }"
+            :class="{ mixed: isMixed(group) }"
+            :style="headingStyle(group)"
           >
             <span>{{ group.typename }} ({{ group.items.length }})</span>
             <span v-if="gridRangeLabel(group)" class="recert-range">{{ gridRangeLabel(group) }}</span>
@@ -105,7 +132,7 @@
               v-for="rec in gridPageItems(group)"
               :key="`grid-${rec.notificationid}-${rec.employeesso}`"
               class="recert-tile"
-              :class="{ 'message-tile': !hasEmployee(rec) }"
+              :class="{ 'message-tile': !hasEmployee(rec), pending: rec.upcoming || rec.resolved }"
               :style="{ '--accent': getTypeColor(rec.typecolor) }"
             >
               <template v-if="hasEmployee(rec)">
@@ -127,6 +154,7 @@
               <!-- No employee: the message IS the tile, so a grid type like
                    Awareness does not render rows of blank placeholder faces. -->
               <div v-else class="recert-message">{{ rec.notification }}</div>
+              <span v-if="stateLabel(rec)" class="state-chip tile-state">{{ stateLabel(rec) }}</span>
             </div>
           </div>
         </section>
@@ -264,7 +292,14 @@ const STYLE_ORDER = { banner: 0, carousel: 1, grid: 2 }
 // row, so same-category types of different styles still render separately.
 const styledGroups = computed(() => {
   const groups = new Map()
-  for (const n of notifications.value.current) {
+  // Upcoming cards belong here too. They used to be filtered into the standard
+  // Upcoming list, which drops anything with a special style - so a scheduled
+  // banner or carousel card rendered nowhere at all until it went live.
+  const tagged = [
+    ...notifications.value.current.map(n => ({ ...n, upcoming: false })),
+    ...notifications.value.upcoming.map(n => ({ ...n, upcoming: true }))
+  ]
+  for (const n of tagged) {
     if (!SPECIAL_STYLES.includes(n.displaystyle)) continue
     const heading = (n.boardcategory || '').trim() || n.typename || n.displaystyle
     const key = `${n.displaystyle}|${heading}`
@@ -282,6 +317,10 @@ const styledGroups = computed(() => {
     }
     groups.get(key).items.push(n)
   }
+  // Live cards lead each row; what has not started yet follows.
+  for (const group of groups.values()) {
+    group.items.sort((a, b) => Number(a.upcoming) - Number(b.upcoming))
+  }
   return [...groups.values()].sort((a, b) =>
     (STYLE_ORDER[a.displaystyle] - STYLE_ORDER[b.displaystyle]) ||
     a.typename.localeCompare(b.typename)
@@ -465,6 +504,29 @@ function getTypeColor(typecolor) {
   return aliases[typecolor] || typecolor || '#14abef'
 }
 
+// What a card's state chip says, or '' while it is simply live. Resolved is the
+// tail a type's grace window buys it; upcoming is scheduled and not yet showing.
+function stateLabel(item) {
+  if (item.upcoming) return `UPCOMING ${formatDateTime(item.starttime)}`
+  return item.resolved ? 'RESOLVED' : ''
+}
+
+// A row holding more than one type: the heading names a category, so it cannot
+// speak for any single type's colour or name.
+function isMixed(group) {
+  return new Set(group.items.map(item => item.typename)).size > 1
+}
+
+// A single-type row wears that type's colour. A mixed one keeps the neutral
+// heading, since picking the first type's colour just mislabels the rest.
+function headingStyle(group) {
+  if (isMixed(group)) return {}
+  return {
+    backgroundColor: getTypeColor(group.typecolor),
+    color: textOn(group.typecolor)
+  }
+}
+
 // Text colour for a heading painted an arbitrary type colour. Gold needs dark
 // text, navy needs white, and a site picks its own hexes - so this is computed
 // rather than a per-colour rule.
@@ -701,6 +763,14 @@ function handlePhotoError(e) {
   overflow: hidden;
 }
 
+/* A row of several types speaks for none of them, so it keeps the neutral
+   heading rather than wearing the first type's colour (or, via the per-style
+   fallbacks above, Recognition's gold and Recertification's blue). */
+.section-title.mixed {
+  background: #4181ff;
+  color: #fff;
+}
+
 .section-title.danger {
   background: #f5365c;
 }
@@ -732,9 +802,75 @@ function handlePhotoError(e) {
 
 .recognition-carousel {
   position: relative;
+  /* Enough for a photo card; a message-only card is shorter and leaves the
+     rest as breathing room rather than resizing the board every 8 seconds. */
   min-height: 170px;
 }
 
+.carousel-pips {
+  display: flex;
+  gap: 8px;
+  justify-content: center;
+  margin-top: 10px;
+}
+.pip {
+  width: 14px;
+  height: 14px;
+  border-radius: 50%;
+  opacity: 0.35;
+  transition: opacity 0.4s ease, transform 0.4s ease;
+}
+.pip.on {
+  opacity: 1;
+  transform: scale(1.25);
+}
+
+.chip-row {
+  display: flex;
+  align-items: center;
+  gap: 10px;
+  margin-bottom: 8px;
+}
+
+/* Scheduled or just-ended cards stay legible but visibly quieter than live
+   ones, so a glance at the board never mistakes one for the other. Carousel
+   cards are dimmed only while ACTIVE: the inactive ones are held at opacity 0
+   by .recognition-card, and a plain .pending rule outranks it, which brought
+   the whole stack back into view on top of each other. */
+.recognition-card.active.pending,
+.recert-tile.pending,
+.banner-strip.pending {
+  opacity: 0.72;
+}
+
+.state-chip {
+  padding: 4px 14px;
+  border-radius: 999px;
+  font-size: 18px;
+  font-weight: 700;
+  letter-spacing: 1px;
+  text-transform: uppercase;
+  background: rgba(255, 255, 255, 0.16);
+  border: 2px solid rgba(255, 255, 255, 0.45);
+  color: #fff;
+  white-space: nowrap;
+}
+.state-chip.tile-state {
+  font-size: 14px;
+  padding: 2px 10px;
+  margin-top: 8px;
+}
+
+.type-chip {
+  padding: 4px 14px;
+  border-radius: 999px;
+  font-size: 20px;
+  font-weight: 700;
+  letter-spacing: 1px;
+  text-transform: uppercase;
+  margin-bottom: 8px;
+}
+
 .recognition-card {
   position: absolute;
   top: 0;
@@ -774,6 +910,8 @@ function handlePhotoError(e) {
 }
 
 .recognition-content {
+  display: flex;
+  flex-direction: column;
   flex: 1;
 }
 
diff --git a/plugins/backups/__init__.py b/plugins/backups/__init__.py
new file mode 100644
index 00000000..a5af74dd
--- /dev/null
+++ b/plugins/backups/__init__.py
@@ -0,0 +1,5 @@
+"""Backups plugin package."""
+
+from .plugin import BackupsPlugin
+
+__all__ = ['BackupsPlugin']
diff --git a/plugins/backups/api/__init__.py b/plugins/backups/api/__init__.py
new file mode 100644
index 00000000..f2d6c9d6
--- /dev/null
+++ b/plugins/backups/api/__init__.py
@@ -0,0 +1,5 @@
+"""Backups plugin API."""
+
+from .routes import backups_bp
+
+__all__ = ['backups_bp']
diff --git a/plugins/backups/api/routes.py b/plugins/backups/api/routes.py
new file mode 100644
index 00000000..deee70b9
--- /dev/null
+++ b/plugins/backups/api/routes.py
@@ -0,0 +1,306 @@
+"""Backups API.
+
+Read + download only. Revisions are created exclusively through the ADR-006
+collector endpoint (POST /api/collector/backups), so there is deliberately no
+create route here: a hand-posted "backup" that never came off a machine would
+poison the history this feature exists to provide.
+
+Downloads differ by storage backend. 'shopdb' kinds re-render from the stored
+projection, which is what lets one revision produce either .reg dialect. 'share'
+kinds are not streamed - ShopDB returns the UNC path for the tech to open,
+because requiring the app server to mount the SFLD share would turn a
+permissions slip into an unexplained empty download.
+"""
+
+from flask import Blueprint, current_app, request, Response
+from flask_jwt_extended import jwt_required
+
+from shopdb.api import (
+    db, Asset,
+    success_response, error_response, ErrorCodes,
+    require_permission,
+)
+
+from ..models import BackupRevision
+from ..services.registry import REGISTRY, getkind
+
+backups_bp = Blueprint('backups', __name__)
+
+
+def _label(revision):
+    """Panel list title: kind-agnostic, readable at a glance."""
+    when = revision.collectedat or revision.createdat
+    stamp = when.strftime('%Y-%m-%d %H:%M') if when else 'unknown time'
+    if revision.sourcefilename:
+        return '{} - {}'.format(stamp, revision.sourcefilename)
+    return stamp
+
+
+def _summary(revision, islatest=False):
+    data = revision.to_dict()
+    data['label'] = _label(revision)
+    data['islatest'] = islatest
+    kind = getkind(revision.backupkind)
+    data['formats'] = kind.formats() if kind else []
+    return data
+
+
+@backups_bp.route('/kinds', methods=['GET'])
+@jwt_required()
+@require_permission('backups.view')
+def list_kinds():
+    """Registered backup kinds, for UI that needs to enumerate them."""
+    return success_response([
+        {
+            'key': kind.key,
+            'displayname': kind.displayname,
+            'storagebackend': kind.storagebackend,
+            'assettypes': list(kind.assettypes),
+            'formats': kind.formats(),
+        }
+        for kind in REGISTRY.values()
+    ])
+
+
+@backups_bp.route('/asset/<int:assetid>', methods=['GET'])
+@jwt_required()
+@require_permission('backups.view')
+def asset_revisions(assetid):
+    """Revision history for one asset, newest first.
+
+    This is the asset-panel endpoint. Optional ?kind= narrows to a single kind,
+    which is how each per-kind panel scopes itself.
+    """
+    asset = db.session.get(Asset, assetid)
+    if asset is None:
+        return error_response(ErrorCodes.NOT_FOUND, 'Asset not found', http_code=404)
+
+    query = db.session.query(BackupRevision).filter(
+        BackupRevision.assetid == assetid)
+
+    kindkey = (request.args.get('kind') or '').strip().lower()
+    if kindkey:
+        if getkind(kindkey) is None:
+            return error_response(ErrorCodes.VALIDATION_ERROR,
+                                  'Unknown kind: {}'.format(kindkey))
+        query = query.filter(BackupRevision.backupkind == kindkey)
+
+    revisions = query.order_by(BackupRevision.backuprevisionid.desc()).all()
+
+    # "Latest" is per kind, not per asset, so an unfiltered listing still marks
+    # the current revision of each kind correctly.
+    seen = set()
+    out = []
+    for revision in revisions:
+        islatest = revision.backupkind not in seen
+        seen.add(revision.backupkind)
+        out.append(_summary(revision, islatest=islatest))
+    return success_response(out)
+
+
+@backups_bp.route('/asset/<int:assetid>/info', methods=['GET'])
+@jwt_required()
+@require_permission('backups.view')
+def asset_info(assetid):
+    """A kind's 'at a glance' card, built from its LATEST revision.
+
+    Generic across kinds: the requested kind owns both the panel declaration
+    and the payload (see BackupKind.infopanel / buildinfo), so a successor to
+    NTLARS/DNC gets its own card without a new endpoint.
+
+    Empty when the machine has no revision of that kind yet, which the generic
+    renderer turns into the panel's empty text.
+    """
+    kindkey = (request.args.get('kind') or 'ntlars').strip().lower()
+    kind = getkind(kindkey)
+    if kind is None:
+        return error_response(ErrorCodes.VALIDATION_ERROR,
+                              'Unknown kind: {}'.format(kindkey))
+
+    revision = (db.session.query(BackupRevision)
+                .filter(BackupRevision.assetid == assetid,
+                        BackupRevision.backupkind == kindkey)
+                .order_by(BackupRevision.backuprevisionid.desc())
+                .first())
+    if revision is None:
+        return success_response({'fields': [], 'sectioncount': 0})
+
+    data = kind.buildinfo(
+        revision.payload, assetid,
+        partmarkertypes=current_app.config.get('BACKUPS_PARTMARKER_TYPES'))
+    data['backuprevisionid'] = revision.backuprevisionid
+    data['collectedat'] = (revision.collectedat.isoformat()
+                           if revision.collectedat else None)
+    return success_response(data)
+
+
+@backups_bp.route('/revisions/<int:backuprevisionid>', methods=['GET'])
+@jwt_required()
+@require_permission('backups.view')
+def get_revision(backuprevisionid):
+    """One revision, including its parsed payload where there is one."""
+    revision = db.session.get(BackupRevision, backuprevisionid)
+    if revision is None:
+        return error_response(ErrorCodes.NOT_FOUND, 'Revision not found',
+                              http_code=404)
+    data = revision.to_dict(includepayload=True)
+    data['label'] = _label(revision)
+    kind = getkind(revision.backupkind)
+    data['formats'] = kind.formats() if kind else []
+    data['displayname'] = kind.displayname if kind else revision.backupkind
+    return success_response(data)
+
+
+@backups_bp.route('/revisions/<int:backuprevisionid>/download', methods=['GET'])
+@jwt_required()
+@require_permission('backups.download')
+def download_revision(backuprevisionid):
+    """Render a revision back to its native file format.
+
+    ?format=ntlars        (default for NTLARS) omits WOW6432Node - the form the
+                          NTLARS Load... button expects
+    ?format=wow6432node   includes WOW6432Node - for `reg import` on 64-bit
+    """
+    revision = db.session.get(BackupRevision, backuprevisionid)
+    if revision is None:
+        return error_response(ErrorCodes.NOT_FOUND, 'Revision not found',
+                              http_code=404)
+
+    kind = getkind(revision.backupkind)
+    if kind is None:
+        return error_response(ErrorCodes.VALIDATION_ERROR,
+                              'Unknown kind: {}'.format(revision.backupkind))
+
+    if revision.storagebackend == 'share':
+        # Not an error: the file exists, ShopDB just is not the one serving it.
+        return success_response({
+            'storagebackend': 'share',
+            'sharepath': revision.sharepath,
+            'sourcefilename': revision.sourcefilename,
+            'message': 'This backup lives on the SFLD share. Open the path directly.',
+        })
+
+    formats = kind.formats()
+    formatid = (request.args.get('format') or '').strip().lower()
+    if not formatid:
+        formatid = formats[0]['id'] if formats else ''
+    if not any(f['id'] == formatid for f in formats):
+        return error_response(
+            ErrorCodes.VALIDATION_ERROR,
+            'Unknown format {!r} for kind {}'.format(formatid, kind.key))
+
+    asset = db.session.get(Asset, revision.assetid)
+    assetnumber = asset.assetnumber if asset else str(revision.assetid)
+    when = revision.collectedat or revision.createdat
+
+    comments = [
+        '{} backup from ShopDB'.format(kind.displayname),
+        'Asset: {}'.format(assetnumber),
+        'Captured: {}'.format(when.strftime('%Y-%m-%d %H:%M:%S') if when else 'unknown'),
+        'Source PC: {}'.format(revision.sourcehostname or 'unknown'),
+        'Revision: {} ({})'.format(revision.backuprevisionid,
+                                   (revision.contenthash or '')[:12]),
+    ]
+
+    try:
+        raw, ext, mimetype = kind.render(revision.payload, formatid,
+                                         comments=comments)
+    except ValueError as exc:
+        return error_response(ErrorCodes.VALIDATION_ERROR, str(exc))
+
+    filename = revision.sourcefilename or '{}-{}{}'.format(
+        assetnumber, kind.key, ext)
+    if formatid == 'wow6432node':
+        filename = '{}-{}-wow6432node{}'.format(assetnumber, kind.key, ext)
+
+    return Response(
+        raw,
+        mimetype=mimetype,
+        headers={'Content-Disposition': 'attachment; filename="{}"'.format(filename)},
+    )
+
+
+@backups_bp.route('/revisions/<int:backuprevisionid>/diff', methods=['GET'])
+@jwt_required()
+@require_permission('backups.view')
+def diff_revision(backuprevisionid):
+    """Diff two revisions of the same asset+kind.
+
+    ?against=<id> picks the comparison revision; default is the immediately
+    preceding one, which answers "what changed" without the user choosing.
+    """
+    revision = db.session.get(BackupRevision, backuprevisionid)
+    if revision is None:
+        return error_response(ErrorCodes.NOT_FOUND, 'Revision not found',
+                              http_code=404)
+    if revision.storagebackend != 'shopdb':
+        return error_response(
+            ErrorCodes.VALIDATION_ERROR,
+            'Kind {} stores opaque files and cannot be diffed'.format(
+                revision.backupkind))
+
+    againstid = request.args.get('against', type=int)
+    if againstid:
+        other = db.session.get(BackupRevision, againstid)
+    else:
+        other = (db.session.query(BackupRevision)
+                 .filter(BackupRevision.assetid == revision.assetid,
+                         BackupRevision.backupkind == revision.backupkind,
+                         BackupRevision.backuprevisionid < revision.backuprevisionid)
+                 .order_by(BackupRevision.backuprevisionid.desc())
+                 .first())
+
+    if other is None:
+        return success_response({
+            'backuprevisionid': revision.backuprevisionid,
+            'againstid': None,
+            'changes': [],
+            'message': 'No earlier revision to compare against.',
+        })
+
+    changes = _diffprojections(other.payload, revision.payload)
+    return success_response({
+        'backuprevisionid': revision.backuprevisionid,
+        'againstid': other.backuprevisionid,
+        'changes': changes,
+        'changecount': len(changes),
+    })
+
+
+def _flatten(projection):
+    """{(subkey, valuename): (type, data)} from a stored projection."""
+    flat = {}
+    for key in (projection or {}).get('keys', []):
+        path = key.get('path', '')
+        for name, entry in (key.get('values') or {}).items():
+            flat[(path, name)] = (entry.get('type'), entry.get('data'))
+    return flat
+
+
+def _diffprojections(old, new):
+    oldflat = _flatten(old)
+    newflat = _flatten(new)
+
+    changes = []
+    for ref in sorted(set(oldflat) | set(newflat)):
+        path, name = ref
+        before = oldflat.get(ref)
+        after = newflat.get(ref)
+        if before == after:
+            continue
+        if before is None:
+            change = 'added'
+        elif after is None:
+            change = 'removed'
+        else:
+            change = 'changed'
+        changes.append({
+            'keypath': path,
+            'valuename': name,
+            'change': change,
+            'before': None if before is None else before[1],
+            'after': None if after is None else after[1],
+            'beforetype': None if before is None else before[0],
+            'aftertype': None if after is None else after[0],
+        })
+    return changes
diff --git a/plugins/backups/frontend/routes.js b/plugins/backups/frontend/routes.js
new file mode 100644
index 00000000..84495279
--- /dev/null
+++ b/plugins/backups/frontend/routes.js
@@ -0,0 +1,21 @@
+/**
+ * Backups plugin routes.
+ *
+ * One route: the per-asset revision history. The asset-detail panel (ADR-010
+ * Path A) lists recent revisions but the generic renderer has no per-item
+ * actions, and downloading a .reg needs a per-revision control plus a dialect
+ * choice. So the panel links here via its manage link rather than the renderer
+ * growing a backups-shaped feature.
+ *
+ * meta.plugin = 'backups' so the ADR-009 guard redirects when the backend
+ * plugin is disabled. requiresAuth because every backups API route is behind
+ * JWT + a backups.* permission; an anonymous visit would only render errors.
+ */
+export default [
+  {
+    path: 'backups/asset/:assetid',
+    name: 'backups-asset-history',
+    component: () => import('./views/BackupHistory.vue'),
+    meta: { requiresAuth: true, plugin: 'backups' }
+  }
+]
diff --git a/plugins/backups/frontend/views/BackupHistory.vue b/plugins/backups/frontend/views/BackupHistory.vue
new file mode 100644
index 00000000..5c1a0f86
--- /dev/null
+++ b/plugins/backups/frontend/views/BackupHistory.vue
@@ -0,0 +1,210 @@
+<template>
+  <div class="bh">
+    <div class="bh-head">
+      <router-link :to="`/assets/${assetid}`" class="bh-back">&larr; Back to asset</router-link>
+      <h1>Configuration Backups</h1>
+      <p v-if="assetname" class="bh-sub">{{ assetname }}</p>
+    </div>
+
+    <div v-if="loading" class="bh-note">Loading...</div>
+    <div v-else-if="error" class="bh-error">{{ error }}</div>
+    <div v-else-if="!revisions.length" class="bh-note">
+      No backups on record for this asset yet.
+    </div>
+
+    <table v-else class="bh-table">
+      <thead>
+        <tr>
+          <th>Captured</th>
+          <th>Kind</th>
+          <th>From PC</th>
+          <th>Hash</th>
+          <th>Changes</th>
+          <th class="bh-actions-col">Download</th>
+        </tr>
+      </thead>
+      <tbody>
+        <tr v-for="rev in revisions" :key="rev.backuprevisionid"
+            :class="{ 'bh-latest': rev.islatest }">
+          <td>
+            {{ formatWhen(rev.collectedat || rev.createdat) }}
+            <span v-if="rev.islatest" class="bh-badge">current</span>
+          </td>
+          <td>{{ rev.backupkind }}</td>
+          <td>{{ rev.sourcehostname || '-' }}</td>
+          <td class="bh-mono">{{ rev.shorthash }}</td>
+          <td>
+            <button v-if="rev.storagebackend === 'shopdb'"
+                    class="bh-link" @click="loadDiff(rev)">
+              {{ diffs[rev.backuprevisionid] ? 'hide' : 'compare' }}
+            </button>
+            <span v-else class="bh-muted">-</span>
+          </td>
+          <td class="bh-actions">
+            <!-- Share-backed revisions are not streamed by ShopDB; show the
+                 UNC path so the tech can open it directly. -->
+            <template v-if="rev.storagebackend === 'share'">
+              <code class="bh-path" :title="rev.sharepath">{{ rev.sharepath }}</code>
+            </template>
+            <template v-else>
+              <button v-for="fmt in rev.formats" :key="fmt.id"
+                      class="bh-btn" :title="fmt.hint"
+                      @click="download(rev, fmt)">
+                {{ fmt.label }}
+              </button>
+            </template>
+          </td>
+        </tr>
+      </tbody>
+    </table>
+
+    <div v-for="rev in revisions" :key="`d-${rev.backuprevisionid}`">
+      <div v-if="diffs[rev.backuprevisionid]" class="bh-diff">
+        <h3>
+          Changes in revision {{ rev.backuprevisionid }}
+          <span v-if="diffs[rev.backuprevisionid].againstid">
+            vs {{ diffs[rev.backuprevisionid].againstid }}
+          </span>
+        </h3>
+        <p v-if="diffs[rev.backuprevisionid].message" class="bh-muted">
+          {{ diffs[rev.backuprevisionid].message }}
+        </p>
+        <table v-else-if="diffs[rev.backuprevisionid].changes.length" class="bh-table">
+          <thead>
+            <tr><th>Key</th><th>Value</th><th>Before</th><th>After</th></tr>
+          </thead>
+          <tbody>
+            <tr v-for="(c, i) in diffs[rev.backuprevisionid].changes" :key="i">
+              <td class="bh-mono">{{ c.keypath || '(root)' }}</td>
+              <td class="bh-mono">{{ c.valuename }}</td>
+              <td class="bh-before">{{ display(c.before) }}</td>
+              <td class="bh-after">{{ display(c.after) }}</td>
+            </tr>
+          </tbody>
+        </table>
+        <p v-else class="bh-muted">No differences.</p>
+      </div>
+    </div>
+  </div>
+</template>
+
+<script setup>
+import { ref, onMounted } from 'vue'
+import { useRoute } from 'vue-router'
+import api from '@/api'
+
+const route = useRoute()
+const assetid = route.params.assetid
+
+const revisions = ref([])
+const diffs = ref({})
+const assetname = ref('')
+const loading = ref(true)
+const error = ref('')
+
+function formatWhen(value) {
+  if (!value) return 'unknown'
+  const d = new Date(value)
+  return isNaN(d) ? value : d.toLocaleString()
+}
+
+function display(value) {
+  if (value === null || value === undefined) return '(absent)'
+  if (value === '') return '(empty)'
+  return String(value)
+}
+
+async function load() {
+  loading.value = true
+  error.value = ''
+  try {
+    const response = await api.get(`/backups/asset/${assetid}`)
+    revisions.value = response.data.data || []
+  } catch (e) {
+    error.value = e.response?.data?.error?.message || 'Failed to load backups.'
+  } finally {
+    loading.value = false
+  }
+  try {
+    const asset = await api.get(`/assets/${assetid}`)
+    assetname.value = asset.data.data?.name || ''
+  } catch (e) {
+    // Name is decoration; the history is still usable without it.
+  }
+}
+
+async function loadDiff(rev) {
+  if (diffs.value[rev.backuprevisionid]) {
+    delete diffs.value[rev.backuprevisionid]
+    diffs.value = { ...diffs.value }
+    return
+  }
+  try {
+    const response = await api.get(
+      `/backups/revisions/${rev.backuprevisionid}/diff`)
+    diffs.value = { ...diffs.value, [rev.backuprevisionid]: response.data.data }
+  } catch (e) {
+    diffs.value = {
+      ...diffs.value,
+      [rev.backuprevisionid]: {
+        changes: [],
+        message: e.response?.data?.error?.message || 'Could not build a diff.'
+      }
+    }
+  }
+}
+
+async function download(rev, fmt) {
+  // responseType blob so the UTF-16LE .reg is not mangled by JSON parsing;
+  // the Content-Disposition filename is not readable cross-origin, so rebuild
+  // it here to match what the server sends.
+  const response = await api.get(
+    `/backups/revisions/${rev.backuprevisionid}/download?format=${fmt.id}`,
+    { responseType: 'blob' })
+  const suffix = fmt.id === 'wow6432node' ? '-wow6432node' : ''
+  const name = `${rev.sourcefilename || rev.backupkind}${suffix}${fmt.ext}`
+    .replace(/(\.reg)+$/, '.reg')
+  const link = document.createElement('a')
+  link.href = URL.createObjectURL(response.data)
+  link.download = name
+  link.click()
+  URL.revokeObjectURL(link.href)
+}
+
+onMounted(load)
+</script>
+
+<style scoped>
+.bh { padding: 1.5rem; }
+.bh-head { margin-bottom: 1rem; }
+.bh-back { font-size: 0.85rem; }
+.bh-sub { color: #666; margin: 0.2rem 0 0; }
+.bh-note, .bh-muted { color: #666; }
+.bh-error { color: #c00; }
+.bh-table { width: 100%; border-collapse: collapse; margin-bottom: 1rem; }
+.bh-table th, .bh-table td {
+  text-align: left; padding: 0.5rem 0.6rem; border-bottom: 1px solid #e0e0e0;
+  vertical-align: top;
+}
+.bh-latest { background: #f4f9ff; }
+.bh-badge {
+  font-size: 0.7rem; background: #0d6efd; color: #fff;
+  border-radius: 3px; padding: 0.05rem 0.35rem; margin-left: 0.4rem;
+}
+.bh-mono, .bh-path { font-family: ui-monospace, Menlo, Consolas, monospace; font-size: 0.82rem; }
+.bh-path { word-break: break-all; }
+.bh-actions-col { width: 22rem; }
+.bh-btn {
+  font-size: 0.78rem; margin: 0 0.3rem 0.3rem 0; padding: 0.25rem 0.5rem;
+  border: 1px solid #0d6efd; background: #fff; color: #0d6efd;
+  border-radius: 4px; cursor: pointer;
+}
+.bh-btn:hover { background: #0d6efd; color: #fff; }
+.bh-link {
+  background: none; border: none; color: #0d6efd; cursor: pointer;
+  padding: 0; font-size: 0.82rem; text-decoration: underline;
+}
+.bh-diff { margin: 0 0 1.5rem; }
+.bh-before { color: #b00; }
+.bh-after { color: #070; }
+</style>
diff --git a/plugins/backups/manifest.json b/plugins/backups/manifest.json
new file mode 100644
index 00000000..55f2806f
--- /dev/null
+++ b/plugins/backups/manifest.json
@@ -0,0 +1,14 @@
+{
+  "name": "backups",
+  "version": "1.0.0",
+  "description": "Per-asset configuration backups with revision history. Kind-pluggable: NTLARS/DNC registry, part marker, UDC, CMM. Parseable kinds store a JSON projection in ShopDB and re-render to their native format on download; opaque vendor formats stay on the SFLD share with ShopDB holding metadata and the UNC pointer.",
+  "author": "ShopDB Team",
+  "dependencies": [],
+  "core_version": ">=0.16.0,<1.0.0",
+  "api_prefix": "/api/backups",
+  "provides": {
+    "features": [
+      "configuration-backups"
+    ]
+  }
+}
diff --git a/plugins/backups/migrations/env.py b/plugins/backups/migrations/env.py
new file mode 100644
index 00000000..a260bd4f
--- /dev/null
+++ b/plugins/backups/migrations/env.py
@@ -0,0 +1,16 @@
+"""Alembic environment for the backups plugin migration chain.
+
+Delegates to the shared runner in shopdb.plugins.alembic_template, which filters
+the metadata to this plugin's tables and drives Alembic against the per-plugin
+version table alembic_version_backups. See ADR-008 for the ownership
+model. Unlike the ten cutover plugins whose 0001 is a no-op anchor, this plugin
+is NEW: its 0001 baseline really CREATES its tables, because the core chain
+never built them.
+"""
+import os
+
+os.environ['PLUGIN_NAME'] = 'backups'
+
+from shopdb.plugins.alembic_template import run_migrations  # noqa: E402
+
+run_migrations()
diff --git a/plugins/backups/migrations/script.py.mako b/plugins/backups/migrations/script.py.mako
new file mode 100644
index 00000000..2c015630
--- /dev/null
+++ b/plugins/backups/migrations/script.py.mako
@@ -0,0 +1,24 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+
+"""
+from alembic import op
+import sqlalchemy as sa
+${imports if imports else ""}
+
+# revision identifiers, used by Alembic.
+revision = ${repr(up_revision)}
+down_revision = ${repr(down_revision)}
+branch_labels = ${repr(branch_labels)}
+depends_on = ${repr(depends_on)}
+
+
+def upgrade():
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade():
+    ${downgrades if downgrades else "pass"}
diff --git a/plugins/backups/migrations/versions/0001_backups_baseline.py b/plugins/backups/migrations/versions/0001_backups_baseline.py
new file mode 100644
index 00000000..885a3ca9
--- /dev/null
+++ b/plugins/backups/migrations/versions/0001_backups_baseline.py
@@ -0,0 +1,61 @@
+"""backups plugin baseline (real create).
+
+Built after the ADR-008 ownership cutover, so unlike the ten cutover plugins
+whose 0001 is a stamp-only anchor, this baseline genuinely CREATES the table.
+The core chain never knew about backuprevisions, so this per-plugin chain is its
+sole authoritative creator.
+
+Emits explicit Alembic ops rather than using the create_plugin_tables helper:
+the helper builds a per-plugin MetaData filtered to the plugin's own tables, so
+the foreign key to the core assets table cannot resolve at CreateTable-compile
+time (NoReferencedTableError). Same shape autogenerate produces. Tables inherit
+the connection's default charset, matching how the core chain creates its own.
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'backups0001baseline'
+down_revision = None
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        'backuprevisions',
+        sa.Column('backuprevisionid', sa.Integer(), nullable=False),
+        sa.Column('assetid', sa.Integer(), nullable=False),
+        sa.Column('backupkind', sa.String(length=50), nullable=False),
+        sa.Column('storagebackend', sa.String(length=20), nullable=False,
+                  server_default='shopdb'),
+        sa.Column('contenthash', sa.String(length=64), nullable=False),
+        sa.Column('payloadjson', sa.Text(length=16777215), nullable=True),
+        sa.Column('sharepath', sa.String(length=500), nullable=True),
+        sa.Column('sourcefilename', sa.String(length=255), nullable=True),
+        sa.Column('bytesize', sa.Integer(), nullable=True),
+        sa.Column('sourcehostname', sa.String(length=255), nullable=True),
+        sa.Column('collectedat', sa.DateTime(), nullable=True),
+        sa.Column('createdat', sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(['assetid'], ['assets.assetid'],
+                                ondelete='CASCADE'),
+        sa.PrimaryKeyConstraint('backuprevisionid'),
+    )
+    op.create_index('ixbackuprevisionsassetid', 'backuprevisions', ['assetid'])
+    op.create_index('ixbackuprevisionsbackupkind', 'backuprevisions',
+                    ['backupkind'])
+    op.create_index('ixbackuprevisionscontenthash', 'backuprevisions',
+                    ['contenthash'])
+    # The dedup read path is "latest revision for this asset+kind", so the
+    # composite index is the one that actually gets used on every collector post.
+    op.create_index('ixbackuprevisionsassetkind', 'backuprevisions',
+                    ['assetid', 'backupkind'])
+
+
+def downgrade():
+    op.drop_index('ixbackuprevisionsassetkind', table_name='backuprevisions')
+    op.drop_index('ixbackuprevisionscontenthash', table_name='backuprevisions')
+    op.drop_index('ixbackuprevisionsbackupkind', table_name='backuprevisions')
+    op.drop_index('ixbackuprevisionsassetid', table_name='backuprevisions')
+    op.drop_table('backuprevisions')
diff --git a/plugins/backups/models/__init__.py b/plugins/backups/models/__init__.py
new file mode 100644
index 00000000..0c5ad86c
--- /dev/null
+++ b/plugins/backups/models/__init__.py
@@ -0,0 +1,5 @@
+"""Backups plugin models."""
+
+from .backup import BackupRevision
+
+__all__ = ['BackupRevision']
diff --git a/plugins/backups/models/backup.py b/plugins/backups/models/backup.py
new file mode 100644
index 00000000..6dfb50e8
--- /dev/null
+++ b/plugins/backups/models/backup.py
@@ -0,0 +1,123 @@
+"""Backup revision model.
+
+One row per DISTINCT configuration snapshot of an asset. The collector runs
+every GE-Enforce cycle across the whole fleet, so the write path dedupes on
+contenthash: a row appears only when a setting actually changed. That is what
+turns a high-frequency collector into a readable revision history.
+
+Two storage backends, chosen by the kind (see services/registry.py):
+
+  'shopdb'  Parsed structured config lives in payloadjson. The original file
+            is not kept because it re-renders exactly from the projection
+            (NTLARS .reg is the motivating case), which also lets one stored
+            revision render into more than one dialect on download.
+
+  'share'   Opaque vendor formats that have no useful JSON representation
+            (part marker files and similar). Bytes stay on the SFLD share and
+            the row carries sharepath plus enough metadata to list, dedupe and
+            link to them. ShopDB never needs to parse these.
+
+contenthash is sha256 over the canonical form of whatever is authoritative for
+the backend: the canonical JSON for 'shopdb', the raw file bytes for 'share'.
+"""
+
+import json
+from datetime import datetime
+
+from shopdb.api import db
+
+
+class BackupRevision(db.Model):
+    """A single point-in-time configuration snapshot of an asset."""
+
+    __tablename__ = 'backuprevisions'
+
+    backuprevisionid = db.Column(db.Integer, primary_key=True)
+
+    # The asset the config BELONGS to, which is not always the asset it was
+    # collected from. NTLARS settings live in the controlling PC's registry but
+    # describe the machine, so the collector reports a machine number and the
+    # kind resolves it to the machine's asset. sourcehostname records the PC it
+    # actually came off.
+    assetid = db.Column(
+        db.Integer,
+        db.ForeignKey('assets.assetid', ondelete='CASCADE'),
+        nullable=False,
+        index=True,
+    )
+
+    backupkind = db.Column(db.String(50), nullable=False, index=True)
+    storagebackend = db.Column(db.String(20), nullable=False, default='shopdb')
+
+    # sha256 of the canonical authoritative form. Dedup key together with
+    # (assetid, backupkind).
+    contenthash = db.Column(db.String(64), nullable=False, index=True)
+
+    # Populated for storagebackend='shopdb' only. MEDIUMTEXT holding serialized
+    # JSON rather than a native JSON column, matching geenforce.manifestjson.
+    # Text also means the bytes come back exactly as written, so the canonical
+    # key ordering the codec produces survives the round trip - a native JSON
+    # column would renormalize it and make diffs between revisions unstable.
+    payloadjson = db.Column(db.Text(length=16777215), nullable=True)
+
+    # Populated for storagebackend='share' only. Full UNC path.
+    sharepath = db.Column(db.String(500), nullable=True)
+
+    # Original filename incl. extension. Vendor tools reject a renamed file, so
+    # downloads hand back exactly this name.
+    sourcefilename = db.Column(db.String(255), nullable=True)
+    bytesize = db.Column(db.Integer, nullable=True)
+
+    # Which PC produced it, and when it was read off that PC (not when ShopDB
+    # stored it - a share-drop fallback can arrive much later).
+    sourcehostname = db.Column(db.String(255), nullable=True)
+    collectedat = db.Column(db.DateTime, nullable=True)
+
+    createdat = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
+
+    __table_args__ = (
+        db.Index('ixbackuprevisionsassetkind', 'assetid', 'backupkind'),
+    )
+
+    @property
+    def payload(self):
+        """Decoded payloadjson, or None for opaque (share-backed) kinds."""
+        if not self.payloadjson:
+            return None
+        try:
+            return json.loads(self.payloadjson)
+        except (ValueError, TypeError) as exc:
+            # Do NOT return None here. A corrupt row would then look identical
+            # to an opaque share-backed revision, and the download route would
+            # fail somewhere further along with an unrelated error. Name the
+            # actual problem and the row it is in.
+            raise ValueError(
+                'backuprevision {} has unreadable payloadjson: {}'.format(
+                    self.backuprevisionid, exc))
+
+    @payload.setter
+    def payload(self, value):
+        if value is None:
+            self.payloadjson = None
+        else:
+            self.payloadjson = json.dumps(value, sort_keys=True,
+                                          separators=(',', ':'))
+
+    def to_dict(self, includepayload=False):
+        data = {
+            'backuprevisionid': self.backuprevisionid,
+            'assetid': self.assetid,
+            'backupkind': self.backupkind,
+            'storagebackend': self.storagebackend,
+            'contenthash': self.contenthash,
+            'shorthash': (self.contenthash or '')[:12],
+            'sharepath': self.sharepath,
+            'sourcefilename': self.sourcefilename,
+            'bytesize': self.bytesize,
+            'sourcehostname': self.sourcehostname,
+            'collectedat': self.collectedat.isoformat() if self.collectedat else None,
+            'createdat': self.createdat.isoformat() if self.createdat else None,
+        }
+        if includepayload:
+            data['payloadjson'] = self.payload
+        return data
diff --git a/plugins/backups/plugin.py b/plugins/backups/plugin.py
new file mode 100644
index 00000000..96a978c6
--- /dev/null
+++ b/plugins/backups/plugin.py
@@ -0,0 +1,332 @@
+"""Backups plugin main class.
+
+Per-asset configuration backups with revision history, pluggable by kind
+(services/registry.py). Accepts collector input over HTTPS per ADR-006 and
+renders one asset panel per kind via the ADR-010 hook.
+
+Dedup is the load-bearing behaviour: GE-Enforce runs the collector every cycle
+across the fleet, so apply_collector_payload inserts a revision only when the
+content hash differs from that asset's latest for that kind. Without it a
+readable history would be buried under millions of identical rows.
+"""
+
+import base64
+import json
+import logging
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Dict, List, Optional, Type
+
+from flask import Flask, Blueprint
+
+from shopdb.plugins.base import BasePlugin, PluginMeta
+
+from .api import backups_bp
+from .models import BackupRevision
+from .services.registry import (REGISTRY, getkind, canonicalhash,
+                               DEFAULTSHAREROOT)
+from .services.retention import prune
+
+logger = logging.getLogger(__name__)
+
+
+class BackupsPlugin(BasePlugin):
+    """Configuration backup + revision history plugin."""
+
+    def __init__(self):
+        self._manifest = self._load_manifest()
+
+    def _load_manifest(self) -> Dict:
+        manifest_path = Path(__file__).parent / 'manifest.json'
+        if manifest_path.exists():
+            with open(manifest_path, 'r') as f:
+                return json.load(f)
+        return {}
+
+    @property
+    def meta(self) -> PluginMeta:
+        return PluginMeta(
+            name=self._manifest.get('name', 'backups'),
+            version=self._manifest.get('version', '1.0.0'),
+            description=self._manifest.get('description', 'Configuration backups'),
+            author=self._manifest.get('author', 'ShopDB Team'),
+            dependencies=self._manifest.get('dependencies', []),
+            core_version=self._manifest.get('core_version', '>=0.2.0,<1.0.0'),
+            api_prefix=self._manifest.get('api_prefix', '/api/backups'),
+        )
+
+    def get_blueprint(self) -> Optional[Blueprint]:
+        return backups_bp
+
+    def get_models(self) -> List[Type]:
+        return [BackupRevision]
+
+    def get_permissions(self) -> List:
+        """RBAC for the human-facing routes.
+
+        Deliberately split: download is separate from view because a .reg is a
+        restorable artifact, so a read-only auditor should be able to see that a
+        machine's config changed without being able to pull a file that
+        reconfigures it.
+
+        No delete permission. Revisions are removed only by the retention
+        policy; a hand-delete route would let someone quietly drop the evidence
+        of a config change, and there is no operational need for it. Collector
+        INGEST is not covered here at all - that path authorizes on a
+        collector.ingest-scoped service token, not a user JWT.
+        """
+        return [
+            ('backups.view', 'View configuration backups', 'backups'),
+            ('backups.download', 'Download configuration backups', 'backups'),
+        ]
+
+    def get_asset_panels(self) -> List[Dict]:
+        """One panel per kind, plus the DNC Info card.
+
+        DNC Info sits ABOVE the history panels (lower position) because it
+        answers the question a tech actually arrives with - what is this
+        machine's controller talking to - while the history panels are for the
+        rarer restore case.
+        """
+        panels = [kind.infopanel() for kind in REGISTRY.values()
+                  if kind.infopanel()]
+        for position, kind in enumerate(REGISTRY.values()):
+            panels.append({
+                'id': 'backups-{}'.format(kind.key),
+                'title': kind.displayname,
+                'assettypes': list(kind.assettypes),
+                'endpoint': '/api/backups/asset/{assetid}?kind=' + kind.key,
+                'render': 'list',
+                'map': {
+                    'title': 'label',
+                    'meta': [
+                        {'key': 'collectedat', 'label': 'Captured', 'format': 'date'},
+                        {'key': 'sourcehostname', 'label': 'From'},
+                        {'key': 'shorthash', 'label': 'Hash', 'mono': True},
+                    ],
+                },
+                'empty': kind.emptytext,
+                # The generic renderer has no per-item actions, and downloading
+                # a .reg needs a per-revision control plus a dialect choice, so
+                # the panel links to the plugin's own history view instead.
+                'manage': {
+                    'to': '/backups/asset/{assetid}',
+                    'label': 'History / download',
+                    'emptylabel': 'View history',
+                },
+                'position': 40 + position,
+            })
+        return panels
+
+    def get_settings_defaults(self) -> List[Dict]:
+        """Schedule + retention, read by the collector and the prune job.
+
+        The interval lives here rather than in the manifest entry because
+        GE-Enforce fires every cycle (several times a day on a shift change).
+        The collector reads backups_intervalhours and no-ops when its marker
+        file is younger than that, so the cadence is changed centrally in
+        ShopDB instead of by editing a script on the share.
+        """
+        return [
+            {
+                'key': 'backups_intervalhours',
+                'value': '24',
+                'valuetype': 'integer',
+                'category': 'backups',
+                'description': 'Minimum hours between collection attempts on a '
+                               'PC. GE-Enforce runs every cycle; the collector '
+                               'skips until this much time has passed.',
+            },
+            {
+                'key': 'backups_shareroot',
+                'value': DEFAULTSHAREROOT,
+                'valuetype': 'string',
+                'category': 'backups',
+                'description': 'UNC root that opaque (non-JSON) backups are '
+                               'written under by the collecting PC. Site-specific.',
+            },
+            {
+                'key': 'backups_retentioncount',
+                'value': '50',
+                'valuetype': 'integer',
+                'category': 'backups',
+                'description': 'Distinct revisions kept per asset per kind. '
+                               '0 keeps everything. The newest and the oldest '
+                               'are never pruned.',
+            },
+            {
+                'key': 'backups_retentiondays',
+                'value': '0',
+                'valuetype': 'integer',
+                'category': 'backups',
+                'description': 'Also prune revisions older than this many days. '
+                               '0 disables age-based pruning.',
+            },
+        ]
+
+    # ---- ADR-006 collector contract -------------------------------------
+
+    def get_collector_schema(self) -> Optional[dict]:
+        return {
+            'identityfield': 'machinenumber',
+            'fields': {
+                'type': 'object',
+                'required': ['machinenumber', 'backupkind'],
+                'properties': {
+                    'machinenumber': {
+                        'type': 'string',
+                        'description': 'Machine number the config belongs to. '
+                                       'Resolved against assets.assetnumber.',
+                    },
+                    'backupkind': {
+                        'type': 'string',
+                        'enum': sorted(REGISTRY.keys()),
+                    },
+                    'contentbase64': {
+                        'type': 'string',
+                        'description': "Raw backup file, base64. Required for "
+                                       "'shopdb' kinds (parsed server-side).",
+                    },
+                    'contenthash': {
+                        'type': 'string',
+                        'description': "sha256 of the file. Required for 'share' "
+                                       "kinds, whose bytes stay on the SFLD share.",
+                    },
+                    'sharepath': {
+                        'type': 'string',
+                        'description': "Full UNC path. Required for 'share' kinds.",
+                    },
+                    'sourcefilename': {'type': 'string'},
+                    'sourcehostname': {
+                        'type': 'string',
+                        'description': 'PC the backup was read from. For NTLARS '
+                                       'this is the controlling PC, not the machine.',
+                    },
+                    'collectedat': {'type': 'string', 'format': 'date-time'},
+                    'bytesize': {'type': 'integer'},
+                },
+            },
+        }
+
+    def apply_collector_payload(self, payload: dict) -> dict:
+        from shopdb.api import db
+
+        warnings = []
+        kindkey = (payload.get('backupkind') or '').strip().lower()
+        kind = getkind(kindkey)
+        if kind is None:
+            raise ValueError('unknown backupkind: {!r}'.format(kindkey))
+
+        assetid, err = kind.resolveassetid(payload)
+        if assetid is None:
+            # Unresolvable is reported, never silently dropped - an unmatched
+            # machine number means a real gap between the fleet and ShopDB.
+            raise ValueError('could not resolve asset: {}'.format(err))
+
+        projection = None
+        sharepath = None
+        sourcefilename = payload.get('sourcefilename')
+        bytesize = payload.get('bytesize')
+
+        if kind.storagebackend == 'shopdb':
+            encoded = payload.get('contentbase64')
+            if not encoded:
+                raise ValueError(
+                    "contentbase64 is required for kind '{}'".format(kindkey))
+            raw = base64.b64decode(encoded)
+            bytesize = bytesize or len(raw)
+            projection = kind.parse(raw)
+            contenthash = canonicalhash(projection)
+
+            # The fleet's machine number (pc-config.txt) is authoritative for
+            # WHICH asset this belongs to, but NTLARS carries its own MachineNo.
+            # A disagreement means the PC is running another machine's config -
+            # worth surfacing rather than quietly filing under the reported one.
+            embedded = getattr(kind, 'embeddedmachineno', None)
+            if embedded is not None:
+                embeddedno = embedded(projection)
+                reported = (payload.get('machinenumber') or '').strip()
+                if embeddedno and reported and embeddedno != reported:
+                    warnings.append(
+                        'reported machine number {!r} but NTLARS is configured '
+                        'for {!r}'.format(reported, embeddedno))
+        else:
+            contenthash = (payload.get('contenthash') or '').strip().lower()
+            sharepath = (payload.get('sharepath') or '').strip()
+            if not contenthash or not sharepath:
+                raise ValueError(
+                    "contenthash and sharepath are required for kind '{}'".format(
+                        kindkey))
+
+        # with_for_update serializes concurrent posts for the same asset+kind.
+        # The whole fleet collects on the same GE-Enforce cycle, so a retry or
+        # two PCs claiming one machine number can otherwise both pass the hash
+        # check and insert duplicate identical revisions. No unique constraint
+        # can cover this: dedup is against the LATEST row only, because a config
+        # reverting to an earlier state is a legitimate new revision.
+        latest = (db.session.query(BackupRevision)
+                  .filter(BackupRevision.assetid == assetid,
+                          BackupRevision.backupkind == kindkey)
+                  .order_by(BackupRevision.backuprevisionid.desc())
+                  .with_for_update()
+                  .first())
+
+        if latest is not None and latest.contenthash == contenthash:
+            return {
+                'action': 'noop',
+                'assetid': assetid,
+                'backuprevisionid': latest.backuprevisionid,
+                'warnings': warnings,
+            }
+
+        collectedat = None
+        rawcollected = payload.get('collectedat')
+        if rawcollected:
+            try:
+                collectedat = datetime.fromisoformat(rawcollected.replace('Z', '+00:00'))
+                if collectedat.tzinfo is not None:
+                    # Convert, don't just drop the offset: createdat is utcnow(),
+                    # so keeping local wall time would skew a -04:00 bay by 4h
+                    # against its own history.
+                    collectedat = collectedat.astimezone(timezone.utc).replace(
+                        tzinfo=None)
+            except ValueError:
+                warnings.append('unparseable collectedat: {!r}'.format(rawcollected))
+
+        revision = BackupRevision(
+            assetid=assetid,
+            backupkind=kindkey,
+            storagebackend=kind.storagebackend,
+            contenthash=contenthash,
+            sharepath=sharepath,
+            sourcefilename=sourcefilename,
+            bytesize=bytesize,
+            sourcehostname=payload.get('sourcehostname'),
+            collectedat=collectedat or datetime.utcnow(),
+        )
+        revision.payload = projection
+        db.session.add(revision)
+        # flush, not commit: the collector dispatcher owns the transaction and
+        # commits after writing its AuditLog row. Committing here would leave an
+        # unaudited revision behind if that write then failed.
+        db.session.flush()
+
+        pruned = prune(
+            assetid, kindkey,
+            retentioncount=self.get_setting('backups_retentioncount', 0),
+            retentiondays=self.get_setting('backups_retentiondays', 0),
+        )
+        if pruned:
+            warnings.append('pruned {} old revision(s) per retention policy'
+                            .format(pruned))
+
+        return {
+            'action': 'created',
+            'assetid': assetid,
+            'backuprevisionid': revision.backuprevisionid,
+            'warnings': warnings,
+        }
+
+    def init_app(self, app: Flask, db_instance) -> None:
+        logger.info('Backups plugin initialized (v%s, kinds: %s)',
+                    self.meta.version, ', '.join(sorted(REGISTRY)))
diff --git a/plugins/backups/services/__init__.py b/plugins/backups/services/__init__.py
new file mode 100644
index 00000000..7563f19e
--- /dev/null
+++ b/plugins/backups/services/__init__.py
@@ -0,0 +1,6 @@
+"""Backups plugin services: kind registry and per-kind codecs."""
+
+from .registry import REGISTRY, BackupKind, getkind, canonicalhash, byteshash, DEFAULTSHAREROOT
+
+__all__ = ['REGISTRY', 'BackupKind', 'getkind', 'canonicalhash', 'byteshash',
+           'DEFAULTSHAREROOT']
diff --git a/plugins/backups/services/dncinfo.py b/plugins/backups/services/dncinfo.py
new file mode 100644
index 00000000..e1333a35
--- /dev/null
+++ b/plugins/backups/services/dncinfo.py
@@ -0,0 +1,161 @@
+"""DNC Info card.
+
+Surfaces the handful of NTLARS settings a tech actually asks about on the
+machine's page, so the common question ("what is this machine's controller
+talking to?") is answered without downloading and reading a .reg.
+
+Sections, drawn from the machine's LATEST ntlars revision:
+
+  eFocas   Fanuc ethernet link - IP, socket, dual-path. Present on 143 of the
+           147 known-good backups, so it is shown whenever it has content.
+  Serial   RS-232 link parameters. Always populated (Baud, Data Bits and
+           friends carry defaults even where the link is unused), so it is
+           always shown.
+  NTSHR    Network share the controller pulls programs from. Populated on only
+           18 of 147, hence the has-content gate: showing an empty NTSHR block
+           on 129 machines would be noise.
+  MARK     Part-marker settings. Gated on the ASSET being a Part Marker in
+           ShopDB, NOT on the key having content.
+
+WHY MARK IS GATED ON THE ASSET, NOT THE KEY:
+    The obvious rule - show MARK when it has content - does not work. MARK is
+    populated on 145 of 147 machines because Baud/Data Bits carry serial
+    defaults everywhere, and the fields that would identify a marker
+    (CageCode, DataHost, DataPath, MarkMasterPath) are empty across the entire
+    corpus. The one field that is set, DncPatterns, reads YES on 103 of 147
+    including ordinary lathes, so it is a DNC pattern-matching option and not a
+    marker flag. No value in the DNC tree distinguishes a part marker, so the
+    machine's type in ShopDB is the only reliable signal.
+"""
+
+# Values equal to these (case-insensitively) count as "no content".
+EMPTYISH = ('', '0', 'no', 'none')
+
+DEFAULTPARTMARKERTYPES = ('Part Marker',)
+
+
+def _keyvalues(projection, path):
+    """Values under one subkey of the stored projection, or {}."""
+    for key in (projection or {}).get('keys', []):
+        if (key.get('path') or '').lower() == path.lower():
+            return key.get('values') or {}
+    return {}
+
+
+def _hascontent(values):
+    """True when at least one value carries something meaningful."""
+    for entry in values.values():
+        data = entry.get('data')
+        if isinstance(data, (list, tuple)):
+            if any(str(x).strip() for x in data):
+                return True
+            continue
+        if str(data).strip().lower() not in EMPTYISH:
+            return True
+    return False
+
+
+def _fields(values, mono=()):
+    """Render a subkey's values as keyvalue panel fields, empties dropped."""
+    out = []
+    for name in sorted(values):
+        data = values[name].get('data')
+        if isinstance(data, (list, tuple)):
+            data = ', '.join(str(x) for x in data)
+        text = '' if data is None else str(data)
+        if not text.strip():
+            continue
+        out.append({
+            'label': name,
+            'value': text,
+            'mono': name in mono,
+        })
+    return out
+
+
+def ispartmarker(assetid, typenames=None):
+    """True when this asset is a Part Marker according to ShopDB.
+
+    Reads the machines plugin defensively: a lean per-site build (ADR-014) may
+    not install it, and the DNC Info card must degrade to "no MARK section"
+    rather than erroring the whole panel.
+    """
+    typenames = tuple(t.lower() for t in (typenames or DEFAULTPARTMARKERTYPES))
+    try:
+        from shopdb.api import db
+        from plugins.machines.models import Machine, MachineType
+    except ImportError:
+        return False
+    try:
+        row = (db.session.query(MachineType.machinetype)
+               .join(Machine, Machine.machinetypeid == MachineType.machinetypeid)
+               .filter(Machine.assetid == assetid)
+               .first())
+    except Exception:
+        return False
+    return bool(row) and (row[0] or '').strip().lower() in typenames
+
+
+def _cncismarker(projection):
+    """True when NTLARS itself says the controller is a marker.
+
+    General\\Cnc reads 'MARKER' on the part markers (0600 and 0614 in the
+    known-good corpus) and a controller family - Fanuc 30, Fanuc 16, OKUMA,
+    Fidia - everywhere else. This is the one place the DNC tree does
+    distinguish a marker; nothing inside the MARK key does, since MARK carries
+    serial defaults on nearly every machine.
+    """
+    general = _keyvalues(projection, 'General')
+    cnc = str((general.get('Cnc') or {}).get('data') or '').strip().upper()
+    return cnc == 'MARKER'
+
+
+def build(projection, assetid, partmarkertypes=None):
+    """Build the DNC Info card payload from a stored ntlars projection.
+
+    Returns {'fields': [...]} in the shape the generic keyvalue renderer wants,
+    with section headings inlined as labelless separators.
+    """
+    sections = []
+
+    # General first: what the controller IS, before what it talks to. Cnc gives
+    # the controller family, NcIF the interface actually in use (EFOCAS on 127
+    # of 147, NTSHR on 16, SERIAL on 3, HSSB on 1), HostType the DNC host.
+    general = _keyvalues(projection, 'General')
+    wanted = ('Cnc', 'NcIF', 'HostType')
+    generalfields = _fields(
+        {n: v for n, v in general.items() if n in wanted})
+    if generalfields:
+        sections.append(('General', generalfields))
+
+    efocas = _keyvalues(projection, 'eFocas')
+    if _hascontent(efocas):
+        sections.append(('eFocas (ethernet link)', _fields(efocas, mono=('IpAddr',))))
+
+    serial = _keyvalues(projection, 'Serial')
+    if serial:
+        sections.append(('Serial (RS-232)', _fields(serial)))
+
+    ntshr = _keyvalues(projection, 'NTSHR')
+    if _hascontent(ntshr):
+        sections.append(('NTSHR (program share)',
+                         _fields(ntshr, mono=('ShrFolder', 'ShrFolder2',
+                                              'ShrFolder3', 'ShrHost'))))
+
+    # Either signal is enough: NTLARS's own Cnc=MARKER works before anyone has
+    # set the machine's type in ShopDB and on lean builds with no machines
+    # plugin, while the ShopDB type still covers a marker whose Cnc says
+    # something else.
+    if _cncismarker(projection) or ispartmarker(assetid, partmarkertypes):
+        mark = _keyvalues(projection, 'MARK')
+        if mark:
+            sections.append(('MARK (part marker)',
+                             _fields(mark, mono=('DataPath', 'MarkMasterPath'))))
+
+    fields = []
+    for title, entries in sections:
+        if not entries:
+            continue
+        fields.append({'label': title, 'value': '', 'heading': True})
+        fields.extend(entries)
+    return {'fields': fields, 'sectioncount': len(sections)}
diff --git a/plugins/backups/services/ntlars.py b/plugins/backups/services/ntlars.py
new file mode 100644
index 00000000..593a75c9
--- /dev/null
+++ b/plugins/backups/services/ntlars.py
@@ -0,0 +1,308 @@
+"""NTLARS / DNC registry backup codec.
+
+Converts between Windows .reg files and a dialect-neutral JSON projection.
+
+WHY DIALECT-NEUTRAL: NTLARS is a 32-bit app, so its settings physically live
+under HKLM\\SOFTWARE\\WOW6432Node\\GE Aircraft Engines\\DNC. But NTLARS's own
+Save... button exports them WITHOUT the WOW6432Node segment (it writes the path
+it asks for, before the WOW64 redirector rewrites it). Both dialects therefore
+exist in the wild:
+
+  NTLARS Save... output      HKLM\\SOFTWARE\\GE Aircraft Engines\\DNC
+  scripted / reg export      HKLM\\SOFTWARE\\WOW6432Node\\GE Aircraft Engines\\DNC
+
+Parsing strips whichever root matched and stores subkeys RELATIVE to it, so the
+stored revision commits to neither. render() then re-attaches whichever root the
+consumer needs:
+
+  dialect='ntlars'      no WOW6432Node - what the NTLARS Load... button expects
+  dialect='wow6432node' explicit - what `reg import` needs on a 64-bit box
+
+Getting this backwards is silent: a reg import of the NTLARS dialect on 64-bit
+writes to the 64-bit hive, where NTLARS will never look, and reports success.
+
+CANONICAL ORDERING: keys and value names are sorted on parse. MySQL's JSON type
+normalizes object key order anyway, so preserving source order is not possible
+end-to-end; sorting makes it deterministic instead, which is what makes diffs
+between revisions stable. Re-rendered files are semantically identical to their
+source, not byte-identical.
+"""
+
+import re
+
+SCHEMA = 'ntlars/1'
+
+REGROOT = 'HKEY_LOCAL_MACHINE'
+DNCPATH = r'SOFTWARE\GE Aircraft Engines\DNC'
+DNCPATHWOW = r'SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC'
+
+ROOTNTLARS = '{}\\{}'.format(REGROOT, DNCPATH)
+ROOTWOW = '{}\\{}'.format(REGROOT, DNCPATHWOW)
+
+# Order is not significant: the two roots diverge immediately after
+# 'SOFTWARE\\' (GE vs WOW), so neither is a string prefix of the other and
+# _striproot's startswith test cannot match the wrong one. Listed longest-first
+# only for readability.
+KNOWNROOTS = (ROOTWOW, ROOTNTLARS)
+
+HEADER = 'Windows Registry Editor Version 5.00'
+
+# hex(N): type codes that appear in .reg files, mapped to registry type names.
+HEXTYPES = {
+    0: 'REG_NONE',
+    1: 'REG_SZ',
+    2: 'REG_EXPAND_SZ',
+    3: 'REG_BINARY',
+    4: 'REG_DWORD',
+    7: 'REG_MULTI_SZ',
+    11: 'REG_QWORD',
+}
+HEXTYPECODES = {v: k for k, v in HEXTYPES.items()}
+
+
+class NtlarsParseError(ValueError):
+    """Raised when input is not a .reg file we can make sense of."""
+
+
+def decodereg(raw):
+    """Decode .reg bytes to text.
+
+    .reg files are conventionally UTF-16LE with a BOM (that is what both
+    regedit and NTLARS emit), but hand-edited ones show up as UTF-8. Sniff the
+    BOM rather than trusting the extension.
+    """
+    if isinstance(raw, str):
+        return raw
+    if raw.startswith(b'\xff\xfe'):
+        return raw.decode('utf-16-le')[1:]
+    if raw.startswith(b'\xfe\xff'):
+        return raw.decode('utf-16-be')[1:]
+    if raw.startswith(b'\xef\xbb\xbf'):
+        return raw.decode('utf-8-sig')
+    # No BOM. UTF-16LE ASCII text has a NUL in every other byte.
+    if b'\x00' in raw[:64]:
+        return raw.decode('utf-16-le', errors='replace')
+    return raw.decode('utf-8', errors='replace')
+
+
+def _unescape(s):
+    return s.replace('\\\\', '\x00').replace('\\"', '"').replace('\x00', '\\')
+
+
+def _escape(s):
+    return s.replace('\\', '\\\\').replace('"', '\\"')
+
+
+def _joincontinuations(text):
+    """Fold .reg line continuations (trailing backslash) into single lines."""
+    out = []
+    for line in text.replace('\r\n', '\n').replace('\r', '\n').split('\n'):
+        if out and out[-1].endswith('\\'):
+            out[-1] = out[-1][:-1] + line.strip()
+        else:
+            out.append(line)
+    return out
+
+
+def _parsehexvalue(body):
+    """Parse the body of a hex:/hex(N): value into (typename, data)."""
+    m = re.match(r'^hex(?:\((?P<code>[0-9a-fA-F]+)\))?:(?P<bytes>.*)$', body, re.S)
+    if not m:
+        raise NtlarsParseError('unparseable hex value: {!r}'.format(body))
+    code = int(m.group('code'), 16) if m.group('code') else 3
+    tokens = [t.strip() for t in m.group('bytes').split(',') if t.strip()]
+    try:
+        data = bytes(int(t, 16) for t in tokens)
+    except ValueError as exc:
+        raise NtlarsParseError('bad hex byte in value: {}'.format(exc))
+
+    typename = HEXTYPES.get(code, 'REG_BINARY')
+
+    # Wide-string hex types decode back to text so diffs stay readable.
+    if typename in ('REG_SZ', 'REG_EXPAND_SZ'):
+        return typename, data.decode('utf-16-le', errors='replace').rstrip('\x00')
+    if typename == 'REG_MULTI_SZ':
+        text = data.decode('utf-16-le', errors='replace')
+        return typename, [p for p in text.split('\x00') if p]
+    # REG_QWORD must come back as an int: _rendervalue turns it back into
+    # little-endian bytes via int(), so storing the "aa,bb" byte form here
+    # would raise at download time - i.e. precisely when someone is trying to
+    # restore a machine.
+    if typename == 'REG_QWORD':
+        return typename, int.from_bytes(data, 'little')
+    return typename, ','.join('{:02x}'.format(b) for b in data)
+
+
+def _parsevalue(body):
+    """Parse the right-hand side of a .reg value assignment."""
+    body = body.strip()
+    if body.startswith('"'):
+        if not body.endswith('"') or len(body) < 2:
+            raise NtlarsParseError('unterminated string value: {!r}'.format(body))
+        return 'REG_SZ', _unescape(body[1:-1])
+    if body.lower().startswith('dword:'):
+        try:
+            return 'REG_DWORD', int(body.split(':', 1)[1].strip(), 16)
+        except ValueError:
+            raise NtlarsParseError('bad dword value: {!r}'.format(body))
+    if body.lower().startswith('hex'):
+        return _parsehexvalue(body)
+    if body == '-':
+        return 'DELETE', None
+    raise NtlarsParseError('unrecognised value form: {!r}'.format(body))
+
+
+def _striproot(keypath):
+    """Strip a known DNC root, returning the relative subkey path.
+
+    Returns None for keys outside the DNC tree so callers can ignore them
+    rather than silently folding unrelated hives into the backup.
+    """
+    for root in KNOWNROOTS:
+        if keypath.upper() == root.upper():
+            return ''
+        prefix = root.upper() + '\\'
+        if keypath.upper().startswith(prefix):
+            return keypath[len(prefix):]
+    return None
+
+
+def parse(raw):
+    """Parse .reg bytes/text into the dialect-neutral JSON projection.
+
+    Returns {'schema', 'sourcedialect', 'keys': [{'path', 'values': {...}}]}
+    with keys and value names sorted for deterministic diffing.
+    """
+    text = decodereg(raw)
+    lines = _joincontinuations(text)
+
+    if not any(line.strip().lower().startswith('windows registry editor')
+               or line.strip().lower().startswith('regedit4')
+               for line in lines[:5]):
+        raise NtlarsParseError('missing "Windows Registry Editor" header')
+
+    keys = {}
+    current = None
+    sawwow = False
+    sawplain = False
+
+    for line in lines:
+        stripped = line.strip()
+        if not stripped or stripped.startswith(';'):
+            continue
+
+        if stripped.startswith('[') and stripped.endswith(']'):
+            keypath = stripped[1:-1].strip()
+            if keypath.startswith('-'):
+                current = None          # key deletion, not a backup concern
+                continue
+            if keypath.upper().startswith(ROOTWOW.upper()):
+                sawwow = True
+            elif keypath.upper().startswith(ROOTNTLARS.upper()):
+                sawplain = True
+            rel = _striproot(keypath)
+            if rel is None:
+                current = None          # outside the DNC tree - ignore
+                continue
+            current = rel
+            keys.setdefault(current, {})
+            continue
+
+        if current is None or '=' not in stripped:
+            continue
+
+        # Match the QUOTED name and split at the '=' that follows its closing
+        # quote. A plain split('=', 1) breaks on any value name containing '='
+        # or an escaped quote - both legal in the registry - and silently drops
+        # the value.
+        match = re.match(r'^(?:@|"((?:[^"\\]|\\.)*)")\s*=\s*(.*)$', stripped, re.S)
+        if not match:
+            continue
+        name = '' if match.group(1) is None else _unescape(match.group(1))
+        body = match.group(2)
+
+        # Deliberately NOT caught. A backup that silently dropped an
+        # unparseable value would present as complete and restore a machine
+        # with a setting missing - the exact silent-failure class this project
+        # has repeatedly been bitten by. Fail the whole parse instead; the
+        # collector reports it and the previous good revision stays newest.
+        typename, data = _parsevalue(body)
+        if typename == 'DELETE':
+            continue
+        keys[current][name] = {'type': typename, 'data': data}
+
+    if not keys:
+        raise NtlarsParseError(
+            'no keys under {} or {} - not an NTLARS DNC backup'.format(
+                ROOTNTLARS, ROOTWOW))
+
+    dialect = 'wow6432node' if sawwow else ('ntlars' if sawplain else 'unknown')
+
+    return {
+        'schema': SCHEMA,
+        'sourcedialect': dialect,
+        'keys': [
+            {'path': path, 'values': dict(sorted(keys[path].items()))}
+            for path in sorted(keys)
+        ],
+    }
+
+
+def _rendervalue(name, entry):
+    typename = entry.get('type', 'REG_SZ')
+    data = entry.get('data')
+    lhs = '@' if name == '' else '"{}"'.format(_escape(name))
+
+    if typename == 'REG_SZ':
+        return '{}="{}"'.format(lhs, _escape('' if data is None else str(data)))
+    if typename == 'REG_DWORD':
+        return '{}=dword:{:08x}'.format(lhs, int(data) & 0xFFFFFFFF)
+    if typename == 'REG_QWORD':
+        return '{}=hex(b):{}'.format(lhs, _hexbytes(
+            int(data).to_bytes(8, 'little')))
+    if typename == 'REG_EXPAND_SZ':
+        payload = ('' if data is None else str(data)).encode('utf-16-le') + b'\x00\x00'
+        return '{}=hex(2):{}'.format(lhs, _hexbytes(payload))
+    if typename == 'REG_MULTI_SZ':
+        parts = data if isinstance(data, list) else [str(data)]
+        payload = ''.join(p + '\x00' for p in parts).encode('utf-16-le') + b'\x00\x00'
+        return '{}=hex(7):{}'.format(lhs, _hexbytes(payload))
+
+    # REG_BINARY / REG_NONE: data is the "aa,bb,cc" form parse() produced.
+    raw = bytes(int(t, 16) for t in str(data).split(',') if t.strip()) if data else b''
+    code = HEXTYPECODES.get(typename, 3)
+    prefix = 'hex:' if code == 3 else 'hex({:x}):'.format(code)
+    return '{}={}{}'.format(lhs, prefix, _hexbytes(raw))
+
+
+def _hexbytes(raw):
+    return ','.join('{:02x}'.format(b) for b in raw)
+
+
+def render(projection, dialect='ntlars', comments=None):
+    """Render the JSON projection back to .reg bytes (UTF-16LE, BOM, CRLF).
+
+    dialect='ntlars'      omits WOW6432Node - use with the NTLARS Load... button
+    dialect='wow6432node' includes it - use with `reg import` on 64-bit
+    """
+    if dialect not in ('ntlars', 'wow6432node'):
+        raise ValueError('unknown dialect: {!r}'.format(dialect))
+    root = ROOTWOW if dialect == 'wow6432node' else ROOTNTLARS
+
+    out = [HEADER, '']
+    for line in (comments or []):
+        out.append('; {}'.format(line))
+    if comments:
+        out.append('')
+
+    for key in projection.get('keys', []):
+        path = key.get('path', '')
+        out.append('[{}]'.format(root + ('\\' + path if path else '')))
+        for name, entry in (key.get('values') or {}).items():
+            out.append(_rendervalue(name, entry))
+        out.append('')
+
+    text = '\r\n'.join(out)
+    if not text.endswith('\r\n'):
+        text += '\r\n'
+    return b'\xff\xfe' + text.encode('utf-16-le')
diff --git a/plugins/backups/services/registry.py b/plugins/backups/services/registry.py
new file mode 100644
index 00000000..449a9071
--- /dev/null
+++ b/plugins/backups/services/registry.py
@@ -0,0 +1,237 @@
+"""Backup kind registry.
+
+A "kind" is one category of configuration backup (NTLARS/DNC registry, part
+marker, UDC, CMM ...). Each kind declares three things the rest of the plugin
+needs and nothing else:
+
+  where its bytes live      'shopdb' (parsed into payloadjson) or 'share'
+                            (opaque vendor file left on the SFLD share)
+  how to resolve an asset   the collector reports an identifier; the kind turns
+                            it into the assetid the revision belongs to
+  how to render downloads   one or more output formats
+
+Adding a kind is a class plus one REGISTRY entry. Nothing else in the plugin
+knows kind names.
+"""
+
+import hashlib
+import json
+
+from . import ntlars as ntlarscodec
+
+# Default root of the opaque-backup tree on the SFLD share. Connected PCs write
+# here directly (they need SFLD creds - a SYSTEM process hitting a UNC path
+# without them gets an access-denied that Test-Path reports as "not found").
+#
+# This is the WEST JEFFERSON path and is only a DEFAULT: the live value is the
+# backups_shareroot setting, because a bundled plugin in a multi-site product
+# must not hardcode one site's file server.
+DEFAULTSHAREROOT = r'\\tsgwp00525.wjs.geaerospace.net\shared\dt\shopfloor\backups'
+
+
+def canonicalhash(projection):
+    """sha256 over the SEMANTIC content only. Dedup key for 'shopdb' kinds.
+
+    Deliberately hashes just schema + keys, excluding sourcedialect. The same
+    settings exported through NTLARS's Save... button and through a scripted
+    reg export parse to identical keys but differing sourcedialect; hashing the
+    whole projection would record that as a change and produce a spurious
+    revision every time the collection route changed. Storage is
+    dialect-neutral, so the dedup key has to be too.
+    """
+    subset = {
+        'schema': projection.get('schema'),
+        'keys': projection.get('keys'),
+    }
+    blob = json.dumps(subset, sort_keys=True, separators=(',', ':'))
+    return hashlib.sha256(blob.encode('utf-8')).hexdigest()
+
+
+def byteshash(raw):
+    """sha256 over raw bytes. Dedup key for 'share' kinds."""
+    return hashlib.sha256(raw).hexdigest()
+
+
+class BackupKind:
+    """Base class. Subclasses override what applies to them."""
+
+    key = None
+    displayname = None
+    storagebackend = 'shopdb'
+    assettypes = ['*']
+    # Human note shown on the panel when there is nothing yet.
+    emptytext = 'No backups on record.'
+
+    def parse(self, raw):
+        """Opaque kinds return None; parseable kinds return the projection."""
+        return None
+
+    def formats(self):
+        """Downloadable formats: [{'id','label','ext','mimetype'}]."""
+        return []
+
+    def render(self, projection, formatid, comments=None):
+        """Return (bytes, extension, mimetype) for a 'shopdb' kind."""
+        raise NotImplementedError
+
+    def resolveassetid(self, payload):
+        """Map a collector payload to the assetid the backup belongs to."""
+        raise NotImplementedError
+
+    def infopanel(self):
+        """Optional 'at a glance' card built from this kind's LATEST revision.
+
+        Returns an ADR-010 panel dict, or None when the kind has no summary
+        worth surfacing. Declared by the kind rather than hardcoded in the
+        plugin so a successor technology (NTLARS/DNC is expected to give way to
+        Shopfloor Connect) ships its own card by adding a class, without
+        touching the plugin or the panel wiring.
+        """
+        return None
+
+    def buildinfo(self, projection, assetid, **options):
+        """Build this kind's info-card payload. Only called when infopanel()."""
+        return {'fields': [], 'sectioncount': 0}
+
+    def sharedir(self, machinetype, identifier, shareroot=None):
+        """UNC directory a 'share' kind's files are expected under.
+
+        Advisory only - the authoritative path is the sharepath the collector
+        reports, since the PC is what actually wrote the file. This builds the
+        conventional location for display and for validating a reported path.
+        """
+        return '{}\\{}\\{}\\{}'.format(
+            shareroot or DEFAULTSHAREROOT,
+            machinetype or 'unknown', identifier or 'unknown', self.key)
+
+
+class NtlarsKind(BackupKind):
+    """NTLARS / DNC registry settings.
+
+    Collected from the CONTROLLING PC's registry but belongs to the MACHINE:
+    the settings describe how to talk to that machine's controller, so they
+    follow the machine when a PC is swapped. resolveassetid therefore keys on
+    the reported machine number, not the hostname.
+    """
+
+    key = 'ntlars'
+    displayname = 'NTLARS / DNC Settings'
+    storagebackend = 'shopdb'
+    assettypes = ['machine']
+    emptytext = 'No NTLARS settings captured yet.'
+
+    @staticmethod
+    def embeddedmachineno(projection):
+        """The MachineNo NTLARS itself is configured with (General tab)."""
+        for key in (projection or {}).get('keys', []):
+            if (key.get('path') or '').lower() == 'general':
+                entry = (key.get('values') or {}).get('MachineNo') or {}
+                return str(entry.get('data') or '').strip()
+        return ''
+
+    def parse(self, raw):
+        """Parse, then refuse to record an unconfigured NTLARS install.
+
+        A freshly imaged PC opens NTLARS with a blank General tab (no MachineNo,
+        CNC, Host or Interface type) until a tech restores the config. Storing
+        that would make an empty config the newest revision at exactly the
+        moment someone needs the last good one - so a blank MachineNo is
+        rejected as "nothing worth backing up" rather than accepted as a change.
+        Two of the 320 known-good backups on the share already have this shape.
+        """
+        projection = ntlarscodec.parse(raw)
+        if not self.embeddedmachineno(projection):
+            raise ValueError(
+                'NTLARS General\\MachineNo is empty - unconfigured install, '
+                'refusing to record it as a revision')
+        return projection
+
+    def formats(self):
+        return [
+            {
+                'id': 'ntlars',
+                'label': 'NTLARS Load... (.reg)',
+                'ext': '.reg',
+                'mimetype': 'application/octet-stream',
+                'hint': 'Restore with the Load... button in the NTLARS settings dialog.',
+            },
+            {
+                'id': 'wow6432node',
+                'label': 'Direct reg import (.reg, WOW6432Node)',
+                'ext': '.reg',
+                'mimetype': 'application/octet-stream',
+                'hint': 'Use when importing outside NTLARS on a 64-bit machine.',
+            },
+        ]
+
+    def render(self, projection, formatid, comments=None):
+        if formatid not in ('ntlars', 'wow6432node'):
+            raise ValueError('unknown format for ntlars kind: {}'.format(formatid))
+        raw = ntlarscodec.render(projection, dialect=formatid, comments=comments)
+        return raw, '.reg', 'application/octet-stream'
+
+    def infopanel(self):
+        return {
+            'id': 'backups-dncinfo',
+            'title': 'DNC Info',
+            'assettypes': ['machine'],
+            'endpoint': '/api/backups/asset/{assetid}/info?kind=ntlars',
+            'render': 'keyvalue',
+            'empty': 'No NTLARS settings captured for this machine yet.',
+            # Above the history panels: this answers the question a tech
+            # arrives with, while history is for the rarer restore case.
+            'position': 38,
+        }
+
+    def buildinfo(self, projection, assetid, **options):
+        from . import dncinfo
+        return dncinfo.build(projection, assetid,
+                             partmarkertypes=options.get('partmarkertypes'))
+
+    def resolveassetid(self, payload):
+        from shopdb.api import db, Asset
+
+        machinenumber = (payload.get('machinenumber') or '').strip()
+        if not machinenumber:
+            return None, 'no machinenumber in payload'
+        asset = db.session.query(Asset).filter(
+            Asset.assetnumber == machinenumber).first()
+        if asset is None:
+            return None, 'no asset with assetnumber {!r}'.format(machinenumber)
+        return asset.assetid, None
+
+
+class PartMarkerKind(BackupKind):
+    """Telesis part marker configuration.
+
+    Opaque vendor filetype with no useful JSON representation, so ShopDB stores
+    metadata and a UNC pointer while the file itself stays on the share. The
+    original filename and extension are preserved because the vendor tool
+    rejects a renamed file.
+    """
+
+    key = 'partmarker'
+    displayname = 'Part Marker Configuration'
+    storagebackend = 'share'
+    assettypes = ['machine', 'measuring_tool']
+    emptytext = 'No part marker backups on record.'
+
+    def resolveassetid(self, payload):
+        from shopdb.api import db, Asset
+
+        identifier = (payload.get('machinenumber')
+                      or payload.get('assetnumber') or '').strip()
+        if not identifier:
+            return None, 'no machinenumber/assetnumber in payload'
+        asset = db.session.query(Asset).filter(
+            Asset.assetnumber == identifier).first()
+        if asset is None:
+            return None, 'no asset with assetnumber {!r}'.format(identifier)
+        return asset.assetid, None
+
+
+REGISTRY = {k.key: k for k in (NtlarsKind(), PartMarkerKind())}
+
+
+def getkind(key):
+    return REGISTRY.get((key or '').strip().lower())
diff --git a/plugins/backups/services/retention.py b/plugins/backups/services/retention.py
new file mode 100644
index 00000000..35abf25d
--- /dev/null
+++ b/plugins/backups/services/retention.py
@@ -0,0 +1,68 @@
+"""Revision retention.
+
+Dedup already keeps growth low - a revision appears only when a setting really
+changed - so retention exists for the pathological case, not the normal one: a
+value that flaps (or two PCs alternately claiming one machine number) would
+otherwise append a revision every collection cycle forever.
+
+Two independent limits, both off by default at 0:
+
+  retentioncount  keep at most N revisions per asset per kind
+  retentiondays   additionally drop anything older than N days
+
+The NEWEST and the OLDEST revision are never pruned. The newest is the one a
+tech restores from; the oldest is the earliest known-good baseline, which is
+usually the most valuable row in the table and the one a naive "keep last N"
+would delete first.
+"""
+
+from datetime import datetime, timedelta
+
+from shopdb.api import db
+
+from ..models import BackupRevision
+
+
+def prune(assetid, backupkind, retentioncount=0, retentiondays=0):
+    """Delete surplus revisions for one asset+kind. Returns the number removed.
+
+    Caller commits. Returns 0 when both limits are disabled.
+    """
+    retentioncount = int(retentioncount or 0)
+    retentiondays = int(retentiondays or 0)
+    if retentioncount <= 0 and retentiondays <= 0:
+        return 0
+
+    revisions = (db.session.query(BackupRevision)
+                 .filter(BackupRevision.assetid == assetid,
+                         BackupRevision.backupkind == backupkind)
+                 .order_by(BackupRevision.backuprevisionid.desc())
+                 .all())
+    if len(revisions) <= 2:
+        return 0                      # newest + oldest are both protected
+
+    newest = revisions[0]
+    oldest = revisions[-1]
+    protected = {newest.backuprevisionid, oldest.backuprevisionid}
+
+    doomed = []
+
+    if retentioncount > 0 and len(revisions) > retentioncount:
+        # Walk from the oldest END of the middle, so the surplus dropped is
+        # always the least recent, never the newest.
+        for revision in revisions[retentioncount:]:
+            if revision.backuprevisionid not in protected:
+                doomed.append(revision)
+
+    if retentiondays > 0:
+        cutoff = datetime.utcnow() - timedelta(days=retentiondays)
+        for revision in revisions:
+            if revision.backuprevisionid in protected:
+                continue
+            stamp = revision.collectedat or revision.createdat
+            if stamp and stamp < cutoff and revision not in doomed:
+                doomed.append(revision)
+
+    for revision in doomed:
+        db.session.delete(revision)
+    return len(doomed)
diff --git a/scripts/import_ntlars_backups.py b/scripts/import_ntlars_backups.py
new file mode 100644
index 00000000..95b69aad
--- /dev/null
+++ b/scripts/import_ntlars_backups.py
@@ -0,0 +1,137 @@
+#!/usr/bin/env python
+"""Seed backup history from a directory of existing NTLARS .reg files.
+
+A site's file share typically already holds per-machine backups collected by
+hand over years, named by machine number. Importing them gives every machine a
+baseline revision on day one, so the feature is useful before the fleet
+collector has run even once.
+
+Files are keyed by machine number in the FILENAME, which is how the share names
+them. The MachineNo embedded in the file is compared against it and any
+disagreement is reported: in the curated set the two always agree, so a
+mismatch means the file is misfiled and should not be trusted as that machine's
+baseline.
+
+Dry run by default. Nothing is written without --commit.
+
+    venv/bin/python scripts/import_ntlars_backups.py /path/to/regs
+    venv/bin/python scripts/import_ntlars_backups.py /path/to/regs --commit
+"""
+
+import argparse
+import os
+import sys
+from datetime import datetime
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from shopdb import create_app                                     # noqa: E402
+from shopdb.api import db                                         # noqa: E402
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('directory', help='directory holding <machinenumber>.reg files')
+    parser.add_argument('--commit', action='store_true',
+                        help='actually write (default is a dry run)')
+    parser.add_argument('--hostname', default='imported',
+                        help="sourcehostname to record (default 'imported')")
+    args = parser.parse_args()
+
+    if not os.path.isdir(args.directory):
+        parser.error('not a directory: {}'.format(args.directory))
+
+    regs = sorted(f for f in os.listdir(args.directory) if f.lower().endswith('.reg'))
+    if not regs:
+        parser.error('no .reg files in {}'.format(args.directory))
+
+    app = create_app()
+    with app.app_context():
+        from shopdb.core.models import Asset
+        from plugins.backups.models import BackupRevision
+        from plugins.backups.services.registry import getkind, canonicalhash
+
+        kind = getkind('ntlars')
+
+        created = skipped = unresolved = rejected = duplicate = 0
+        mismatches = []
+
+        for filename in regs:
+            path = os.path.join(args.directory, filename)
+            machinenumber = os.path.splitext(filename)[0]
+
+            with open(path, 'rb') as handle:
+                raw = handle.read()
+
+            try:
+                projection = kind.parse(raw)
+            except ValueError as exc:
+                # Unconfigured NTLARS installs are in the corpus; they are not
+                # a usable baseline for anyone.
+                print('  REJECT  {:<12} {}'.format(machinenumber, exc))
+                rejected += 1
+                continue
+
+            embedded = kind.embeddedmachineno(projection)
+            if embedded and embedded != machinenumber:
+                mismatches.append((machinenumber, embedded))
+
+            asset = db.session.query(Asset).filter(
+                Asset.assetnumber == machinenumber).first()
+            if asset is None:
+                print('  NOASSET {:<12} no asset with that assetnumber'.format(
+                    machinenumber))
+                unresolved += 1
+                continue
+
+            contenthash = canonicalhash(projection)
+            latest = (db.session.query(BackupRevision)
+                      .filter(BackupRevision.assetid == asset.assetid,
+                              BackupRevision.backupkind == 'ntlars')
+                      .order_by(BackupRevision.backuprevisionid.desc())
+                      .first())
+            if latest is not None and latest.contenthash == contenthash:
+                duplicate += 1
+                continue
+
+            if args.commit:
+                revision = BackupRevision(
+                    assetid=asset.assetid,
+                    backupkind='ntlars',
+                    storagebackend='shopdb',
+                    contenthash=contenthash,
+                    sourcefilename=filename,
+                    bytesize=len(raw),
+                    sourcehostname=args.hostname,
+                    # The file's mtime is the closest thing to when the config
+                    # was actually captured; createdat records the import.
+                    collectedat=datetime.utcfromtimestamp(os.path.getmtime(path)),
+                )
+                revision.payload = projection
+                db.session.add(revision)
+            created += 1
+
+        if args.commit:
+            db.session.commit()
+
+        print()
+        print('files            : {}'.format(len(regs)))
+        print('would create     : {}'.format(created) if not args.commit
+              else 'created          : {}'.format(created))
+        print('already current  : {}'.format(duplicate))
+        print('no matching asset: {}'.format(unresolved))
+        print('rejected (blank) : {}'.format(rejected))
+        if mismatches:
+            print()
+            print('filename/embedded MachineNo disagreements ({}):'.format(
+                len(mismatches)))
+            for filename_no, embedded_no in mismatches[:20]:
+                print('  file={:<10} embedded={}'.format(filename_no, embedded_no))
+            print('  These are likely misfiled; review before trusting them.')
+        if not args.commit:
+            print()
+            print('DRY RUN - nothing written. Re-run with --commit to apply.')
+
+
+if __name__ == '__main__':
+    main()
diff --git a/shopdb/plugins/alembic_template.py b/shopdb/plugins/alembic_template.py
index 213a15c3..9f9cacaf 100644
--- a/shopdb/plugins/alembic_template.py
+++ b/shopdb/plugins/alembic_template.py
@@ -47,6 +47,7 @@ logger = logging.getLogger('alembic.env.plugin')
 # Explicit table-ownership map. Adding tables to a plugin requires updating
 # this dict so the per-plugin migration knows which tables to include.
 PLUGIN_TABLE_OWNERS: dict[str, Iterable[str]] = {
+    'backups':       ('backuprevisions',),
     'computers':     ('computertypes', 'computers', 'computerinstalledapps',
                       'accessprotocols', 'computeraccess'),
     'employees':     ('directoryemployees',),
diff --git a/tests/test_plugin_migrations.py b/tests/test_plugin_migrations.py
index 5acbd2e3..a00b0643 100644
--- a/tests/test_plugin_migrations.py
+++ b/tests/test_plugin_migrations.py
@@ -46,6 +46,8 @@ CUTOVER_PLUGINS = (
 # stamp '<plugin>0001anchor'; measuringtools stamps its real baseline id.
 EXPECTED_HEAD_REVISION = {plugin: f'{plugin}0001anchor' for plugin in CUTOVER_PLUGINS}
 EXPECTED_HEAD_REVISION['measuringtools'] = 'measuringtools0001baseline'
+# backups is also post-cutover: its 0001 really creates backuprevisions.
+EXPECTED_HEAD_REVISION['backups'] = 'backups0001baseline'
 # geenforce adds the content-addressed blob store (manifestblobs) on top of its
 # baseline.
 EXPECTED_HEAD_REVISION['geenforce'] = 'geenforce0002blobs'
diff --git a/tests/test_plugins/test_backups.py b/tests/test_plugins/test_backups.py
new file mode 100644
index 00000000..55d11599
--- /dev/null
+++ b/tests/test_plugins/test_backups.py
@@ -0,0 +1,633 @@
+"""Tests for the backups plugin.
+
+Two layers:
+
+  Pure codec tests exercise the NTLARS .reg <-> JSON round trip and the dialect
+  toggle. These need no app and are where the real risk lives: getting the
+  WOW6432Node dialect wrong is silent - a reg import of the NTLARS dialect on a
+  64-bit box writes to a hive NTLARS never reads and still reports success.
+
+  Collector tests cover the dedup rule that makes revision history usable.
+  GE-Enforce runs the collector every cycle across the fleet, so an unchanged
+  machine must produce 'noop', not another row.
+"""
+
+import base64
+import json
+
+import pytest
+
+from plugins.backups.services import ntlars, registry
+
+
+# A minimal but representative NTLARS export: the root key plus one subkey,
+# both value types that actually occur in the 320 real backups (REG_SZ and
+# REG_DWORD), written in the WOW6432Node dialect that scripted exports produce.
+SAMPLEREG = (
+    'Windows Registry Editor Version 5.00\r\n'
+    '\r\n'
+    '; NTLARS DNC Registry Backup\r\n'
+    '\r\n'
+    r'[HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC]' '\r\n'
+    '"COMPUTERNAME"="GGBX0NH3ESF"\r\n'
+    '\r\n'
+    r'[HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC\Btr]' '\r\n'
+    '"BTR Rate"="300"\r\n'
+    '"Auto Rewind"="YES"\r\n'
+    '"CmntLag"=dword:00000000\r\n'
+)
+
+
+def _asbytes(text):
+    """Encode as UTF-16LE with a BOM, which is what regedit and NTLARS emit."""
+    return b'\xff\xfe' + text.encode('utf-16-le')
+
+
+# =============================================================================
+# Codec: parsing
+# =============================================================================
+
+def test_parse_reads_utf16_with_bom():
+    projection = ntlars.parse(_asbytes(SAMPLEREG))
+    assert projection['schema'] == ntlars.SCHEMA
+    assert [k['path'] for k in projection['keys']] == ['', 'Btr']
+
+
+def test_parse_reads_utf8_without_bom():
+    projection = ntlars.parse(SAMPLEREG.encode('utf-8'))
+    assert [k['path'] for k in projection['keys']] == ['', 'Btr']
+
+
+def test_parse_strips_the_root_so_storage_is_dialect_neutral():
+    """Both dialects must parse to the same projection - that is the point."""
+    wow = ntlars.parse(_asbytes(SAMPLEREG))
+    plain = ntlars.parse(_asbytes(
+        SAMPLEREG.replace(r'SOFTWARE\WOW6432Node\GE', r'SOFTWARE\GE')))
+    assert wow['keys'] == plain['keys']
+    assert wow['sourcedialect'] == 'wow6432node'
+    assert plain['sourcedialect'] == 'ntlars'
+
+
+def test_parse_records_value_types():
+    projection = ntlars.parse(_asbytes(SAMPLEREG))
+    btr = next(k for k in projection['keys'] if k['path'] == 'Btr')
+    assert btr['values']['BTR Rate'] == {'type': 'REG_SZ', 'data': '300'}
+    assert btr['values']['CmntLag'] == {'type': 'REG_DWORD', 'data': 0}
+
+
+def test_parse_sorts_keys_and_values_for_stable_diffs():
+    projection = ntlars.parse(_asbytes(SAMPLEREG))
+    btr = next(k for k in projection['keys'] if k['path'] == 'Btr')
+    assert list(btr['values']) == sorted(btr['values'])
+
+
+def test_parse_ignores_keys_outside_the_dnc_tree():
+    """An unrelated hive in the same file must not be folded into the backup."""
+    polluted = SAMPLEREG + (
+        r'[HKEY_LOCAL_MACHINE\SOFTWARE\Some Other Vendor\Thing]' '\r\n'
+        '"Nope"="should not appear"\r\n'
+    )
+    projection = ntlars.parse(_asbytes(polluted))
+    allvalues = {name for k in projection['keys'] for name in k['values']}
+    assert 'Nope' not in allvalues
+
+
+def test_parse_rejects_a_file_with_no_registry_header():
+    with pytest.raises(ntlars.NtlarsParseError):
+        ntlars.parse(b'this is not a reg file')
+
+
+def test_parse_rejects_a_reg_file_with_no_dnc_keys():
+    other = (
+        'Windows Registry Editor Version 5.00\r\n\r\n'
+        r'[HKEY_LOCAL_MACHINE\SOFTWARE\Unrelated]' '\r\n'
+        '"X"="1"\r\n'
+    )
+    with pytest.raises(ntlars.NtlarsParseError):
+        ntlars.parse(_asbytes(other))
+
+
+# =============================================================================
+# Codec: rendering and the dialect toggle
+# =============================================================================
+
+def test_render_ntlars_dialect_omits_wow6432node():
+    """This is the form the NTLARS Load... button expects."""
+    projection = ntlars.parse(_asbytes(SAMPLEREG))
+    text = ntlars.render(projection, dialect='ntlars').decode('utf-16-le')
+    assert r'SOFTWARE\GE Aircraft Engines\DNC' in text
+    assert 'WOW6432Node' not in text
+
+
+def test_render_wow6432node_dialect_includes_it():
+    """This is the form `reg import` needs on a 64-bit machine."""
+    projection = ntlars.parse(_asbytes(SAMPLEREG))
+    text = ntlars.render(projection, dialect='wow6432node').decode('utf-16-le')
+    assert r'SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC' in text
+
+
+def test_render_emits_utf16le_with_bom_and_crlf():
+    projection = ntlars.parse(_asbytes(SAMPLEREG))
+    raw = ntlars.render(projection)
+    assert raw.startswith(b'\xff\xfe')
+    assert '\r\n' in raw.decode('utf-16-le')
+
+
+def test_render_starts_with_the_registry_header():
+    projection = ntlars.parse(_asbytes(SAMPLEREG))
+    text = ntlars.render(projection).decode('utf-16-le')
+    assert text.lstrip('\ufeff').startswith(ntlars.HEADER)
+
+
+def test_render_rejects_an_unknown_dialect():
+    projection = ntlars.parse(_asbytes(SAMPLEREG))
+    with pytest.raises(ValueError):
+        ntlars.render(projection, dialect='nonsense')
+
+
+def test_render_includes_comments_when_given():
+    projection = ntlars.parse(_asbytes(SAMPLEREG))
+    text = ntlars.render(projection, comments=['machine 3204']).decode('utf-16-le')
+    assert '; machine 3204' in text
+
+
+@pytest.mark.parametrize('dialect', ['ntlars', 'wow6432node'])
+def test_roundtrip_is_lossless_through_both_dialects(dialect):
+    first = ntlars.parse(_asbytes(SAMPLEREG))
+    second = ntlars.parse(ntlars.render(first, dialect=dialect))
+    assert first['keys'] == second['keys']
+
+
+def test_dword_survives_the_roundtrip_as_a_dword():
+    """A REG_DWORD silently becoming REG_SZ would restore a broken config."""
+    first = ntlars.parse(_asbytes(SAMPLEREG))
+    second = ntlars.parse(ntlars.render(first))
+    btr = next(k for k in second['keys'] if k['path'] == 'Btr')
+    assert btr['values']['CmntLag']['type'] == 'REG_DWORD'
+
+
+# =============================================================================
+# Hashing / dedup key
+# =============================================================================
+
+def test_canonicalhash_is_stable_across_key_order():
+    a = {'keys': [{'path': '', 'values': {'A': {'type': 'REG_SZ', 'data': '1'}}}]}
+    b = json.loads(json.dumps(a))
+    assert registry.canonicalhash(a) == registry.canonicalhash(b)
+
+
+def test_canonicalhash_changes_when_a_value_changes():
+    projection = ntlars.parse(_asbytes(SAMPLEREG))
+    before = registry.canonicalhash(projection)
+    changed = ntlars.parse(_asbytes(SAMPLEREG.replace('"300"', '"600"')))
+    assert registry.canonicalhash(changed) != before
+
+
+# =============================================================================
+# Kind registry
+# =============================================================================
+
+def test_ntlars_kind_stores_in_shopdb_and_offers_both_dialects():
+    kind = registry.getkind('ntlars')
+    assert kind.storagebackend == 'shopdb'
+    assert {f['id'] for f in kind.formats()} == {'ntlars', 'wow6432node'}
+
+
+def test_ntlars_default_format_is_the_load_button_dialect():
+    """First format wins when the caller does not choose, so order matters."""
+    assert registry.getkind('ntlars').formats()[0]['id'] == 'ntlars'
+
+
+def test_partmarker_kind_is_share_backed_and_not_renderable():
+    kind = registry.getkind('partmarker')
+    assert kind.storagebackend == 'share'
+    assert kind.formats() == []
+    assert kind.parse(b'anything') is None
+
+
+def test_partmarker_sharedir_is_under_the_sfld_backups_root():
+    path = registry.getkind('partmarker').sharedir('lathe', '3204')
+    assert path.startswith(registry.DEFAULTSHAREROOT)
+    assert path.endswith(r'lathe\3204\partmarker')
+
+
+def test_getkind_is_case_insensitive_and_returns_none_when_unknown():
+    assert registry.getkind('NTLARS') is not None
+    assert registry.getkind('nosuchkind') is None
+
+
+# =============================================================================
+# Blank-config guard
+# =============================================================================
+
+BLANKREG = (
+    'Windows Registry Editor Version 5.00\r\n\r\n'
+    r'[HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC\General]' '\r\n'
+    '"Cnc"=""\r\n'
+    '"MachineNo"=""\r\n'
+    '"HostType"=""\r\n'
+)
+
+CONFIGUREDREG = BLANKREG.replace('"MachineNo"=""', '"MachineNo"="3204"')
+
+
+def test_kind_rejects_an_unconfigured_ntlars_install():
+    """A freshly imaged PC must not overwrite good history with a blank config."""
+    with pytest.raises(ValueError, match='MachineNo'):
+        registry.getkind('ntlars').parse(_asbytes(BLANKREG))
+
+
+def test_kind_accepts_a_configured_install():
+    projection = registry.getkind('ntlars').parse(_asbytes(CONFIGUREDREG))
+    assert registry.NtlarsKind.embeddedmachineno(projection) == '3204'
+
+
+def test_embeddedmachineno_is_empty_when_general_is_absent():
+    projection = ntlars.parse(_asbytes(SAMPLEREG))
+    assert registry.NtlarsKind.embeddedmachineno(projection) == ''
+
+
+# =============================================================================
+# Collector path (apply_collector_payload) - the dedup rule that makes revision
+# history usable. Needs an app context for the DB.
+# =============================================================================
+
+from shopdb import create_app                                    # noqa: E402
+from shopdb.extensions import db as _db                          # noqa: E402
+from shopdb.plugins import plugin_manager                        # noqa: E402
+
+
+@pytest.fixture(scope='module')
+def bk_app():
+    """Testing app with the backups plugin registered, plus one machine asset."""
+    saved = (plugin_manager._app, plugin_manager._db, plugin_manager.registry,
+             plugin_manager.loader, plugin_manager.migration_manager,
+             plugin_manager._registered_prefixes)
+    application = create_app('testing')
+
+    from plugins.backups.plugin import BackupsPlugin
+    plugin = BackupsPlugin()
+    pm = application.extensions['plugin_manager']
+    if plugin.meta.api_prefix not in pm._registered_prefixes:
+        pm._register_plugin_components(plugin)
+
+    with application.app_context():
+        _db.create_all()
+        from shopdb.core.models import AssetType, AssetStatus, Asset
+        assettype = AssetType(assettype='machine', description='Machine')
+        status = AssetStatus(status='In Use', description='In use')
+        _db.session.add_all([assettype, status])
+        _db.session.flush()
+        # assetnumber is the machine number the collector reports.
+        _db.session.add(Asset(assetnumber='3204', name='Machine 3204',
+                              assettypeid=assettype.assettypeid,
+                              statusid=status.statusid))
+        _db.session.commit()
+        yield application
+        _db.session.remove()
+        _db.drop_all()
+
+    (plugin_manager._app, plugin_manager._db, plugin_manager.registry,
+     plugin_manager.loader, plugin_manager.migration_manager,
+     plugin_manager._registered_prefixes) = saved
+
+
+@pytest.fixture
+def bk_plugin(bk_app):
+    from plugins.backups.plugin import BackupsPlugin
+    from plugins.backups.models import BackupRevision
+    with bk_app.app_context():
+        _db.session.query(BackupRevision).delete()
+        _db.session.commit()
+        yield BackupsPlugin()
+
+
+def _payload(reg=CONFIGUREDREG, **over):
+    data = {
+        'machinenumber': '3204',
+        'backupkind': 'ntlars',
+        'contentbase64': base64.b64encode(_asbytes(reg)).decode('ascii'),
+        'sourcehostname': 'GGBX0NH3ESF',
+    }
+    data.update(over)
+    return data
+
+
+def test_first_post_creates_a_revision(bk_app, bk_plugin):
+    with bk_app.app_context():
+        result = bk_plugin.apply_collector_payload(_payload())
+        assert result['action'] == 'created'
+        assert result['backuprevisionid']
+
+
+def test_identical_repost_is_a_noop_not_a_new_revision(bk_app, bk_plugin):
+    """GE-Enforce reposts every cycle; unchanged machines must not add rows."""
+    with bk_app.app_context():
+        first = bk_plugin.apply_collector_payload(_payload())
+        second = bk_plugin.apply_collector_payload(_payload())
+        assert second['action'] == 'noop'
+        assert second['backuprevisionid'] == first['backuprevisionid']
+
+
+def test_changed_settings_create_a_second_revision(bk_app, bk_plugin):
+    with bk_app.app_context():
+        bk_plugin.apply_collector_payload(_payload())
+        changed = CONFIGUREDREG.replace('"Cnc"=""', '"Cnc"="OKUMA"')
+        result = bk_plugin.apply_collector_payload(_payload(reg=changed))
+        assert result['action'] == 'created'
+
+
+def test_the_same_config_in_the_other_dialect_is_a_noop(bk_app, bk_plugin):
+    """Dedup must be dialect-neutral or the collection route causes churn."""
+    with bk_app.app_context():
+        bk_plugin.apply_collector_payload(_payload())
+        plain = CONFIGUREDREG.replace(r'SOFTWARE\WOW6432Node\GE', r'SOFTWARE\GE')
+        result = bk_plugin.apply_collector_payload(_payload(reg=plain))
+        assert result['action'] == 'noop'
+
+
+def test_blank_config_is_rejected_through_the_collector(bk_app, bk_plugin):
+    with bk_app.app_context():
+        with pytest.raises(ValueError, match='MachineNo'):
+            bk_plugin.apply_collector_payload(_payload(reg=BLANKREG))
+
+
+def test_unresolvable_machine_number_raises_rather_than_dropping(bk_app, bk_plugin):
+    with bk_app.app_context():
+        with pytest.raises(ValueError, match='resolve'):
+            bk_plugin.apply_collector_payload(_payload(machinenumber='9999'))
+
+
+def test_unknown_kind_raises(bk_app, bk_plugin):
+    with bk_app.app_context():
+        with pytest.raises(ValueError, match='unknown backupkind'):
+            bk_plugin.apply_collector_payload(_payload(backupkind='nope'))
+
+
+def test_machinenumber_disagreement_is_warned_not_hidden(bk_app, bk_plugin):
+    """PC configured for another machine must surface, not file silently."""
+    with bk_app.app_context():
+        other = CONFIGUREDREG.replace('"MachineNo"="3204"', '"MachineNo"="7602"')
+        result = bk_plugin.apply_collector_payload(_payload(reg=other))
+        assert result['action'] == 'created'
+        assert any('7602' in w for w in result['warnings'])
+
+
+def test_revision_attaches_to_the_machine_not_the_reporting_pc(bk_app, bk_plugin):
+    """The whole point: history survives replacement of the controlling PC."""
+    from shopdb.core.models import Asset
+    from plugins.backups.models import BackupRevision
+    with bk_app.app_context():
+        bk_plugin.apply_collector_payload(_payload())
+        revision = _db.session.query(BackupRevision).first()
+        machine = _db.session.get(Asset, revision.assetid)
+        assert machine.assetnumber == '3204'
+        assert revision.sourcehostname == 'GGBX0NH3ESF'
+
+
+def test_a_replacement_pc_continues_the_same_history(bk_app, bk_plugin):
+    from plugins.backups.models import BackupRevision
+    with bk_app.app_context():
+        bk_plugin.apply_collector_payload(_payload())
+        changed = CONFIGUREDREG.replace('"Cnc"=""', '"Cnc"="OKUMA"')
+        bk_plugin.apply_collector_payload(
+            _payload(reg=changed, sourcehostname='NEWPC001'))
+        revisions = _db.session.query(BackupRevision).all()
+        assert len(revisions) == 2
+        assert {r.assetid for r in revisions} == {revisions[0].assetid}
+        assert {r.sourcehostname for r in revisions} == {'GGBX0NH3ESF', 'NEWPC001'}
+
+
+def test_share_kind_requires_contenthash_and_sharepath(bk_app, bk_plugin):
+    with bk_app.app_context():
+        with pytest.raises(ValueError, match='contenthash and sharepath'):
+            bk_plugin.apply_collector_payload({
+                'machinenumber': '3204', 'backupkind': 'partmarker'})
+
+
+def test_share_kind_stores_the_pointer_without_a_payload(bk_app, bk_plugin):
+    from plugins.backups.models import BackupRevision
+    with bk_app.app_context():
+        result = bk_plugin.apply_collector_payload({
+            'machinenumber': '3204',
+            'backupkind': 'partmarker',
+            'contenthash': 'a' * 64,
+            'sharepath': r'\\server\share\backups\lathe\3204\partmarker\x.tpm',
+            'sourcefilename': 'x.tpm',
+        })
+        assert result['action'] == 'created'
+        revision = _db.session.get(BackupRevision, result['backuprevisionid'])
+        assert revision.storagebackend == 'share'
+        assert revision.payload is None
+        assert revision.sourcefilename == 'x.tpm'
+
+
+def test_collectedat_offset_is_converted_to_utc_not_truncated(bk_app, bk_plugin):
+    from plugins.backups.models import BackupRevision
+    with bk_app.app_context():
+        result = bk_plugin.apply_collector_payload(
+            _payload(collectedat='2026-08-07T08:00:00-04:00'))
+        revision = _db.session.get(BackupRevision, result['backuprevisionid'])
+        assert revision.collectedat.hour == 12
+
+
+# =============================================================================
+# Retention
+# =============================================================================
+
+def _makerevisions(assetid, count):
+    from plugins.backups.models import BackupRevision
+    for i in range(count):
+        revision = BackupRevision(
+            assetid=assetid, backupkind='ntlars', storagebackend='shopdb',
+            contenthash='{:064d}'.format(i))
+        revision.payload = {'schema': ntlars.SCHEMA, 'keys': []}
+        _db.session.add(revision)
+    _db.session.commit()
+
+
+def test_prune_is_a_noop_when_both_limits_are_disabled(bk_app, bk_plugin):
+    from plugins.backups.services.retention import prune
+    from plugins.backups.models import BackupRevision
+    from shopdb.core.models import Asset
+    with bk_app.app_context():
+        assetid = _db.session.query(Asset).first().assetid
+        _makerevisions(assetid, 10)
+        assert prune(assetid, 'ntlars', 0, 0) == 0
+        assert _db.session.query(BackupRevision).count() == 10
+
+
+def test_prune_keeps_the_configured_count(bk_app, bk_plugin):
+    from plugins.backups.services.retention import prune
+    from plugins.backups.models import BackupRevision
+    from shopdb.core.models import Asset
+    with bk_app.app_context():
+        assetid = _db.session.query(Asset).first().assetid
+        _makerevisions(assetid, 10)
+        removed = prune(assetid, 'ntlars', retentioncount=5)
+        _db.session.commit()
+        # 10 -> 5 newest, but the oldest is protected, so 6 survive.
+        assert removed == 4
+        assert _db.session.query(BackupRevision).count() == 6
+
+
+def test_prune_never_deletes_the_newest_or_the_oldest(bk_app, bk_plugin):
+    """The newest is what a tech restores; the oldest is the baseline."""
+    from plugins.backups.services.retention import prune
+    from plugins.backups.models import BackupRevision
+    from shopdb.core.models import Asset
+    with bk_app.app_context():
+        assetid = _db.session.query(Asset).first().assetid
+        _makerevisions(assetid, 10)
+        ids = [r.backuprevisionid for r in
+               _db.session.query(BackupRevision)
+               .order_by(BackupRevision.backuprevisionid).all()]
+        prune(assetid, 'ntlars', retentioncount=1)
+        _db.session.commit()
+        surviving = {r.backuprevisionid for r in
+                     _db.session.query(BackupRevision).all()}
+        assert ids[0] in surviving and ids[-1] in surviving
+
+
+def test_prune_leaves_two_or_fewer_revisions_alone(bk_app, bk_plugin):
+    from plugins.backups.services.retention import prune
+    from shopdb.core.models import Asset
+    with bk_app.app_context():
+        assetid = _db.session.query(Asset).first().assetid
+        _makerevisions(assetid, 2)
+        assert prune(assetid, 'ntlars', retentioncount=1) == 0
+
+
+# =============================================================================
+# DNC Info card
+# =============================================================================
+
+from plugins.backups.services import dncinfo                       # noqa: E402
+
+DNCINFOREG = (
+    'Windows Registry Editor Version 5.00\r\n\r\n'
+    r'[HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC\General]' '\r\n'
+    '"MachineNo"="3204"\r\n'
+    '"Cnc"="OKUMA"\r\n'
+    '"NcIF"="EFOCAS"\r\n'
+    '"HostType"="WILM"\r\n\r\n'
+    r'[HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC\eFocas]' '\r\n'
+    '"IpAddr"="192.168.1.1"\r\n'
+    '"SocketNo"="8193"\r\n\r\n'
+    r'[HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC\Serial]' '\r\n'
+    '"Baud"="9600"\r\n\r\n'
+    r'[HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC\NTSHR]' '\r\n'
+    '"ShrHost"=""\r\n'
+    '"ShrFolder"=""\r\n\r\n'
+    r'[HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC\MARK]' '\r\n'
+    '"Baud"="9600"\r\n'
+    '"DncPatterns"="YES"\r\n'
+)
+
+
+def _headings(card):
+    return [f['label'] for f in card['fields'] if f.get('heading')]
+
+
+def test_dncinfo_shows_efocas_and_serial():
+    card = dncinfo.build(ntlars.parse(_asbytes(DNCINFOREG)), assetid=0)
+    assert any('eFocas' in h for h in _headings(card))
+    assert any('Serial' in h for h in _headings(card))
+
+
+def test_dncinfo_hides_ntshr_when_it_has_no_content():
+    """NTSHR is populated on only 18 of 147 machines; empty blocks are noise."""
+    card = dncinfo.build(ntlars.parse(_asbytes(DNCINFOREG)), assetid=0)
+    assert not any('NTSHR' in h for h in _headings(card))
+
+
+def test_dncinfo_shows_ntshr_when_populated():
+    populated = DNCINFOREG.replace('"ShrHost"=""', '"ShrHost"="WJFMS3"')
+    card = dncinfo.build(ntlars.parse(_asbytes(populated)), assetid=0)
+    assert any('NTSHR' in h for h in _headings(card))
+
+
+def test_dncinfo_hides_mark_for_a_machine_that_is_not_a_part_marker():
+    """MARK carries serial defaults on 145 of 147 machines, so presence of the
+    key must NOT be what reveals the section."""
+    card = dncinfo.build(ntlars.parse(_asbytes(DNCINFOREG)), assetid=0)
+    assert not any('MARK' in h for h in _headings(card))
+
+
+def test_dncinfo_shows_mark_when_the_asset_is_a_part_marker(monkeypatch):
+    monkeypatch.setattr(dncinfo, 'ispartmarker', lambda *a, **k: True)
+    card = dncinfo.build(ntlars.parse(_asbytes(DNCINFOREG)), assetid=0)
+    assert any('MARK' in h for h in _headings(card))
+
+
+def test_dncinfo_drops_empty_values_within_a_shown_section():
+    populated = DNCINFOREG.replace('"ShrHost"=""', '"ShrHost"="WJFMS3"')
+    card = dncinfo.build(ntlars.parse(_asbytes(populated)), assetid=0)
+    labels = [f['label'] for f in card['fields'] if not f.get('heading')]
+    assert 'ShrHost' in labels
+    assert 'ShrFolder' not in labels
+
+
+def test_dncinfo_is_empty_for_a_projection_with_nothing_interesting():
+    card = dncinfo.build(ntlars.parse(_asbytes(CONFIGUREDREG)), assetid=0)
+    assert card['sectioncount'] == 0
+    assert card['fields'] == []
+
+
+def test_ispartmarker_is_false_when_the_machines_plugin_is_absent(monkeypatch):
+    """Lean builds (ADR-014) may omit machines; the card must degrade, not error."""
+    import builtins
+    real = builtins.__import__
+
+    def fake(name, *args, **kwargs):
+        if name.startswith('plugins.machines'):
+            raise ImportError('not installed')
+        return real(name, *args, **kwargs)
+
+    monkeypatch.setattr(builtins, '__import__', fake)
+    assert dncinfo.ispartmarker(1) is False
+
+
+def test_infopanel_is_kind_owned_so_a_successor_ships_its_own_card():
+    """NTLARS/DNC is expected to give way to Shopfloor Connect; a new kind must
+    be able to add its card without touching the plugin or the endpoint."""
+    ntlarskind = registry.getkind('ntlars')
+    panel = ntlarskind.infopanel()
+    assert panel['id'] == 'backups-dncinfo'
+    assert 'kind=ntlars' in panel['endpoint']
+    # A kind with nothing to summarise contributes no panel at all.
+    assert registry.getkind('partmarker').infopanel() is None
+
+
+def test_base_kind_buildinfo_is_an_empty_card():
+    assert registry.BackupKind().buildinfo({}, 0) == {'fields': [], 'sectioncount': 0}
+
+
+def test_dncinfo_general_section_leads_with_controller_identity():
+    card = dncinfo.build(ntlars.parse(_asbytes(DNCINFOREG)), assetid=0)
+    assert _headings(card)[0] == 'General'
+    labels = [f['label'] for f in card['fields'] if not f.get('heading')]
+    assert 'Cnc' in labels and 'HostType' in labels
+
+
+def test_dncinfo_general_omits_the_rest_of_the_key():
+    """General carries 23 values; the card shows only the three that matter."""
+    reg = DNCINFOREG.replace('"MachineNo"="3204"',
+                             '"MachineNo"="3204"\r\n"Debug"="NO"\r\n"Site"="WJ"')
+    card = dncinfo.build(ntlars.parse(_asbytes(reg)), assetid=0)
+    labels = [f['label'] for f in card['fields'] if not f.get('heading')]
+    assert 'Debug' not in labels and 'Site' not in labels
+
+
+def test_cnc_marker_reveals_the_mark_section_without_shopdb():
+    """General\\Cnc='MARKER' is the one DNC-side part-marker signal, and works
+    before anyone sets the machine type in ShopDB."""
+    reg = DNCINFOREG.replace('"Cnc"="OKUMA"', '"Cnc"="MARKER"')
+    card = dncinfo.build(ntlars.parse(_asbytes(reg)), assetid=0)
+    assert any('MARK' in h for h in _headings(card))
+
+
+def test_ordinary_controller_does_not_reveal_mark():
+    reg = DNCINFOREG  # fixture already carries Cnc=OKUMA
+    card = dncinfo.build(ntlars.parse(_asbytes(reg)), assetid=0)
+    assert not any('MARK' in h for h in _headings(card))