Add GE-Enforce observed-state reporting: receipt + self-heal from PCs

PCs now report enforcement results back to shopdb, closing the desired-vs-observed
loop.

- POST /api/geenforce/report (geenforce.report service token): each cycle a PC
  posts the published version it applied, install/skip/fail/filtered counts, and
  per-entry outcomes.
- Two tables: manifestenforcementreports (latest-per-host + history: applied
  version, enforcer version, counts, derived status ok/selfhealed/failed) and
  manifestenforcementresults (per entry: action installed/skipped/failed,
  selfhealed flag, exit code, warning/error message).
- RECEIVED: reports carry the applied version; the admin view derives
  receivedlatest by comparing it to the scope's current published version, so
  the fleet view shows which PCs picked up an update.
- SELF-HEAL: per-entry action captures drift correction (installed when it
  should already be present) vs skipped (already good) vs failed, with messages.
- Admin reads: GET /reports (fleet compliance rollup) and GET /reports/<id>
  (per-entry detail). New geenforce.report permission.
- Tables added to the (undeployed) 0001 baseline; geenforce.post_report is a
  service-token endpoint so it is exempt from the JWT authz sweep, like the
  collector blueprint. 8 reporting tests; full suite green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

plugins/geenforce/api/routes.py

@@ -19,26 +19,36 @@ from shopdb.api import (
     service_token_authorized,
 )
 
-from ..models import ManifestScope, ManifestPublishedVersion
+from ..models import (
+    ManifestScope, ManifestPublishedVersion, ManifestEnforcementReport,
+)
 from ..serializer import scope_to_manifest
+from .. import service
 
 geenforce_bp = Blueprint('geenforce', __name__)
 
 FETCH_SCOPE = 'geenforce.fetch'
+REPORT_SCOPE = 'geenforce.report'
 
 
-def require_fetch_token(f):
-    """Require a geenforce.fetch service token OR the env bootstrap key."""
-    @wraps(f)
-    def decorated(*args, **kwargs):
-        if service_token_authorized(FETCH_SCOPE):
-            return f(*args, **kwargs)
-        expected = current_app.config.get('GEENFORCE_API_KEY')
-        if expected and request.headers.get('X-API-Key') == expected:
-            return f(*args, **kwargs)
-        return error_response(ErrorCodes.UNAUTHORIZED, 'Invalid API key',
-                              http_code=401)
-    return decorated
+def _require_service_token(scope):
+    """Decorator factory: require `scope` service token OR the env bootstrap key."""
+    def wrapper(f):
+        @wraps(f)
+        def decorated(*args, **kwargs):
+            if service_token_authorized(scope):
+                return f(*args, **kwargs)
+            expected = current_app.config.get('GEENFORCE_API_KEY')
+            if expected and request.headers.get('X-API-Key') == expected:
+                return f(*args, **kwargs)
+            return error_response(ErrorCodes.UNAUTHORIZED, 'Invalid API key',
+                                  http_code=401)
+        return decorated
+    return wrapper
+
+
+require_fetch_token = _require_service_token(FETCH_SCOPE)
+require_report_token = _require_service_token(REPORT_SCOPE)
 
 
 # -- client-facing endpoint ---------------------------------------------------
@@ -78,6 +88,28 @@ def get_manifest():
                              'X-Manifest-Version': str(published.versionnumber)})
 
 
+# -- client reporting (observed state) ----------------------------------------
+
+@geenforce_bp.route('/report', methods=['POST'])
+@require_report_token
+def post_report():
+    """Record one PC's enforcement cycle: applied version (did it receive the
+    update?) + per-entry self-heal outcomes (installed/skipped/failed)."""
+    payload = request.get_json(silent=True) or {}
+    if not (payload.get('hostname') or '').strip():
+        return error_response(ErrorCodes.VALIDATION_ERROR,
+                              'hostname is required', http_code=400)
+    try:
+        report = service.record_enforcement_report(payload)
+        db.session.commit()
+    except ValueError as exc:
+        db.session.rollback()
+        return error_response(ErrorCodes.VALIDATION_ERROR, str(exc),
+                              http_code=400)
+    return success_response({'reportid': report.reportid,
+                             'status': report.status}, message='recorded')
+
+
 # -- admin read surface (P2 adds full CRUD + publish) -------------------------
 
 @geenforce_bp.route('/scopes', methods=['GET'])
@@ -115,3 +147,92 @@ def preview_scope(scopeid):
                               http_code=404)
     return success_response({'scopename': scope.scopename,
                              'manifest': scope_to_manifest(scope)})
+
+
+def _current_published_version(scopename, phase):
+    scope = ManifestScope.query.filter_by(
+        scopename=scopename, phase=phase).first()
+    if not scope:
+        return None
+    published = scope.publishedversions.filter_by(iscurrent=True).first()
+    return published.versionnumber if published else None
+
+
+@geenforce_bp.route('/reports', methods=['GET'])
+@jwt_required()
+@require_permission('geenforce.manage')
+def list_reports():
+    """Latest enforcement report per PC (fleet compliance view).
+
+    Each row shows the applied vs latest published version (receivedlatest = the
+    PC picked up the update) and the install/skip/fail counts + status.
+    Optional filters: hostname, scopename.
+    """
+    query = ManifestEnforcementReport.query.filter_by(iscurrent=True)
+    hostname = request.args.get('hostname')
+    scopename = request.args.get('scopename')
+    if hostname:
+        query = query.filter(ManifestEnforcementReport.hostname.ilike(hostname))
+    if scopename:
+        query = query.filter_by(scopename=scopename)
+    reports = query.order_by(
+        ManifestEnforcementReport.receivedat.desc()).all()
+
+    latest_cache = {}
+    data = []
+    for report in reports:
+        key = (report.scopename, report.phase)
+        if key not in latest_cache:
+            latest_cache[key] = _current_published_version(*key)
+        latest = latest_cache[key]
+        data.append({
+            'reportid': report.reportid,
+            'hostname': report.hostname,
+            'scopename': report.scopename,
+            'phase': report.phase,
+            'appliedversion': report.appliedversion,
+            'latestversion': latest,
+            'receivedlatest': (latest is not None
+                               and report.appliedversion == latest),
+            'enforcerversion': report.enforcerversion,
+            'status': report.status,
+            'installed': report.installedcount,
+            'skipped': report.skippedcount,
+            'failed': report.failedcount,
+            'filtered': report.filteredcount,
+            'lastcheckin': (report.lastcheckin.isoformat() + 'Z'
+                            if report.lastcheckin else None),
+            'receivedat': (report.receivedat.isoformat() + 'Z'
+                           if report.receivedat else None),
+        })
+    return success_response(data)
+
+
+@geenforce_bp.route('/reports/<int:reportid>', methods=['GET'])
+@jwt_required()
+@require_permission('geenforce.manage')
+def get_report(reportid):
+    """One report with per-entry outcomes (self-heal / failure detail)."""
+    report = db.session.get(ManifestEnforcementReport, reportid)
+    if not report:
+        return error_response(ErrorCodes.NOT_FOUND, 'No such report',
+                              http_code=404)
+    latest = _current_published_version(report.scopename, report.phase)
+    return success_response({
+        'reportid': report.reportid,
+        'hostname': report.hostname,
+        'scopename': report.scopename,
+        'phase': report.phase,
+        'appliedversion': report.appliedversion,
+        'latestversion': latest,
+        'receivedlatest': (latest is not None
+                           and report.appliedversion == latest),
+        'status': report.status,
+        'results': [{
+            'entryname': r.entryname,
+            'action': r.action,
+            'selfhealed': r.selfhealed,
+            'exitcode': r.exitcode,
+            'message': r.message,
+        } for r in report.results],
+    })

plugins/geenforce/migrations/versions/0001_geenforce_baseline.py

@@ -173,6 +173,46 @@ def upgrade():
         sa.PrimaryKeyConstraint('payloadid'),
     )
     op.create_index('idx_payload_entry', 'manifestpayloads', ['entryid'])
+    op.create_table(
+        'manifestenforcementreports',
+        sa.Column('reportid', sa.Integer(), nullable=False),
+        sa.Column('hostname', sa.String(length=100), nullable=False),
+        sa.Column('scopename', sa.String(length=64), nullable=False),
+        sa.Column('phase', sa.String(length=16), nullable=False),
+        sa.Column('appliedversion', sa.Integer(), nullable=True),
+        sa.Column('enforcerversion', sa.String(length=20), nullable=True),
+        sa.Column('installedcount', sa.Integer(), nullable=False),
+        sa.Column('skippedcount', sa.Integer(), nullable=False),
+        sa.Column('failedcount', sa.Integer(), nullable=False),
+        sa.Column('filteredcount', sa.Integer(), nullable=False),
+        sa.Column('status', sa.String(length=16), nullable=False),
+        sa.Column('lastcheckin', sa.DateTime(), nullable=True),
+        sa.Column('receivedat', sa.DateTime(), nullable=False),
+        sa.Column('iscurrent', sa.Boolean(), nullable=False),
+        sa.PrimaryKeyConstraint('reportid'),
+    )
+    op.create_index('idx_report_host', 'manifestenforcementreports',
+                    ['hostname'])
+    op.create_index('idx_report_current', 'manifestenforcementreports',
+                    ['iscurrent'])
+    op.create_index('idx_report_host_scope', 'manifestenforcementreports',
+                    ['hostname', 'scopename', 'phase'])
+    op.create_table(
+        'manifestenforcementresults',
+        sa.Column('resultid', sa.Integer(), nullable=False),
+        sa.Column('reportid', sa.Integer(), nullable=False),
+        sa.Column('entryname', sa.String(length=128), nullable=False),
+        sa.Column('action', sa.String(length=16), nullable=False),
+        sa.Column('selfhealed', sa.Boolean(), nullable=False),
+        sa.Column('exitcode', sa.Integer(), nullable=True),
+        sa.Column('message', sa.Text(), nullable=True),
+        sa.ForeignKeyConstraint(['reportid'],
+                                ['manifestenforcementreports.reportid'],
+                                ondelete='CASCADE'),
+        sa.PrimaryKeyConstraint('resultid'),
+    )
+    op.create_index('idx_result_report', 'manifestenforcementresults',
+                    ['reportid'])
     op.create_table(
         'pctypealiases',
         sa.Column('aliasid', sa.Integer(), nullable=False),
@@ -187,6 +227,8 @@ def upgrade():
 
 def downgrade():
     op.drop_table('pctypealiases')
+    op.drop_table('manifestenforcementresults')
+    op.drop_table('manifestenforcementreports')
     op.drop_table('manifestpayloads')
     op.drop_table('manifestpublishedversions')
     op.drop_table('manifestinusecheckprocesses')

plugins/geenforce/models/__init__.py

@@ -10,6 +10,8 @@ from .manifest import (
     ManifestInUseCheckProcess,
     ManifestPublishedVersion,
     ManifestPayload,
+    ManifestEnforcementReport,
+    ManifestEnforcementResult,
     PcTypeAlias,
     PHASES,
     ENTRY_TYPES,
@@ -30,6 +32,8 @@ __all__ = [
     'ManifestInUseCheckProcess',
     'ManifestPublishedVersion',
     'ManifestPayload',
+    'ManifestEnforcementReport',
+    'ManifestEnforcementResult',
     'PcTypeAlias',
     'PHASES',
     'ENTRY_TYPES',

plugins/geenforce/models/manifest.py

@@ -271,6 +271,66 @@ class ManifestPayload(db.Model):
     uploadedat = db.Column(db.DateTime, nullable=False)
 
 
+class ManifestEnforcementReport(db.Model):
+    """One enforcement cycle reported by a PC (observed state).
+
+    Each cycle the client POSTs its result for a scope: the published version it
+    actually applied (so shopdb knows whether the PC RECEIVED the latest update),
+    the enforcer version, and the installed/skipped/failed/filtered counts. The
+    latest report per (hostname, scopename, phase) carries `iscurrent`; older
+    ones are history. Pairs desired state (the manifest) with observed state.
+    """
+    __tablename__ = 'manifestenforcementreports'
+
+    reportid = db.Column(db.Integer, primary_key=True)
+    hostname = db.Column(db.String(100), nullable=False, index=True)
+    scopename = db.Column(db.String(64), nullable=False)
+    phase = db.Column(db.String(16), nullable=False, default='runtime')
+    # Published version the client actually ran; compare to the scope's current
+    # published version to see whether this PC received the latest manifest.
+    appliedversion = db.Column(db.Integer, nullable=True)
+    enforcerversion = db.Column(db.String(20), nullable=True)
+    installedcount = db.Column(db.Integer, nullable=False, default=0)
+    skippedcount = db.Column(db.Integer, nullable=False, default=0)
+    failedcount = db.Column(db.Integer, nullable=False, default=0)
+    filteredcount = db.Column(db.Integer, nullable=False, default=0)
+    # 'ok' | 'failed' (any failure) | 'selfhealed' (drift corrected, no failure).
+    status = db.Column(db.String(16), nullable=False, default='ok')
+    lastcheckin = db.Column(db.DateTime, nullable=True)   # PC-reported time
+    receivedat = db.Column(db.DateTime, nullable=False)   # server time
+    iscurrent = db.Column(db.Boolean, nullable=False, default=True, index=True)
+
+    results = db.relationship(
+        'ManifestEnforcementResult', back_populates='report',
+        cascade='all, delete-orphan', lazy='selectin')
+
+    __table_args__ = (
+        db.Index('idx_report_host_scope', 'hostname', 'scopename', 'phase'),
+    )
+
+
+class ManifestEnforcementResult(db.Model):
+    """One entry's outcome within an enforcement cycle (self-heal detail)."""
+    __tablename__ = 'manifestenforcementresults'
+
+    resultid = db.Column(db.Integer, primary_key=True)
+    reportid = db.Column(
+        db.Integer,
+        db.ForeignKey('manifestenforcementreports.reportid', ondelete='CASCADE'),
+        nullable=False, index=True)
+    entryname = db.Column(db.String(128), nullable=False)
+    # 'installed' (action fired - a self-heal when it should already be present),
+    # 'skipped' (detected present), 'failed', 'filtered'.
+    action = db.Column(db.String(16), nullable=False)
+    # True when this install was a drift correction (self-heal), not a first
+    # install. Client-supplied; defaults to whether the action installed.
+    selfhealed = db.Column(db.Boolean, nullable=False, default=False)
+    exitcode = db.Column(db.Integer, nullable=True)
+    message = db.Column(db.Text, nullable=True)   # warning / error text
+
+    report = db.relationship('ManifestEnforcementReport', back_populates='results')
+
+
 class PcTypeAlias(db.Model):
     """Mirror of the engine lib's PCTypes alias graph (Install-FromManifest.ps1).
 

plugins/geenforce/plugin.py

@@ -21,7 +21,8 @@ from .api import geenforce_bp
 from .models import (
     ManifestScope, ManifestEntry, ManifestEntryPcType, ManifestEntryHostname,
     ManifestEntryMachineNumber, ManifestInUseCheck, ManifestInUseCheckProcess,
-    ManifestPublishedVersion, ManifestPayload, PcTypeAlias,
+    ManifestPublishedVersion, ManifestPayload, ManifestEnforcementReport,
+    ManifestEnforcementResult, PcTypeAlias,
 )
 from .filters import ALIAS_GROUPS
 
@@ -61,7 +62,8 @@ class GeEnforcePlugin(BasePlugin):
             ManifestScope, ManifestEntry, ManifestEntryPcType,
             ManifestEntryHostname, ManifestEntryMachineNumber,
             ManifestInUseCheck, ManifestInUseCheckProcess,
-            ManifestPublishedVersion, ManifestPayload, PcTypeAlias,
+            ManifestPublishedVersion, ManifestPayload,
+            ManifestEnforcementReport, ManifestEnforcementResult, PcTypeAlias,
         ]
 
     def get_permissions(self) -> List:
@@ -73,6 +75,8 @@ class GeEnforcePlugin(BasePlugin):
              'geenforce'),
             ('geenforce.fetch', 'Fetch published manifests (client service token)',
              'geenforce'),
+            ('geenforce.report', 'Report enforcement results (client service token)',
+             'geenforce'),
         ]
 
     def init_app(self, app: Flask, db_instance) -> None:

plugins/geenforce/service.py

@@ -14,7 +14,10 @@ from sqlalchemy import func
 
 from shopdb.api import db
 
-from .models import ManifestScope, ManifestPublishedVersion
+from .models import (
+    ManifestScope, ManifestPublishedVersion, ManifestEnforcementReport,
+    ManifestEnforcementResult,
+)
 from .importer import build_entry
 from .serializer import scope_to_json
 
@@ -90,6 +93,76 @@ def rollback_scope(scopename, phase, versionnumber):
     return versionnumber
 
 
+def record_enforcement_report(payload):
+    """Record one PC's enforcement cycle (observed state). Upserts the latest
+    report per (hostname, scopename, phase) and keeps prior ones as history.
+
+    Payload (all but hostname optional):
+      hostname, scopename, phase, appliedversion, enforcerversion, lastcheckin,
+      counts {installed, skipped, failed, filtered},
+      results [{name, action, selfhealed, exitcode, message}]
+
+    Returns the new ManifestEnforcementReport (uncommitted).
+    """
+    hostname = (payload.get('hostname') or '').strip()
+    if not hostname:
+        raise ValueError('hostname is required')
+    scopename = (payload.get('scopename') or '').strip()
+    phase = (payload.get('phase') or 'runtime').strip()
+    counts = payload.get('counts') or {}
+    results = payload.get('results') or []
+
+    failed = int(counts.get('failed', 0))
+    installed = int(counts.get('installed', 0))
+    # Derive status: any failure wins; else drift-corrected installs = selfhealed.
+    if failed > 0:
+        status = 'failed'
+    elif installed > 0 or any(r.get('selfhealed') for r in results):
+        status = 'selfhealed'
+    else:
+        status = 'ok'
+
+    # Demote the prior current report for this host+scope.
+    ManifestEnforcementReport.query.filter_by(
+        hostname=hostname, scopename=scopename, phase=phase, iscurrent=True
+    ).update({'iscurrent': False})
+
+    report = ManifestEnforcementReport(
+        hostname=hostname,
+        scopename=scopename,
+        phase=phase,
+        appliedversion=payload.get('appliedversion'),
+        enforcerversion=payload.get('enforcerversion'),
+        installedcount=installed,
+        skippedcount=int(counts.get('skipped', 0)),
+        failedcount=failed,
+        filteredcount=int(counts.get('filtered', 0)),
+        status=status,
+        lastcheckin=_parse_dt(payload.get('lastcheckin')),
+        receivedat=_utcnow(),
+        iscurrent=True)
+    for item in results:
+        report.results.append(ManifestEnforcementResult(
+            entryname=item.get('name', ''),
+            action=item.get('action', ''),
+            selfhealed=bool(item.get('selfhealed',
+                                     item.get('action') == 'installed')),
+            exitcode=item.get('exitcode'),
+            message=item.get('message')))
+    db.session.add(report)
+    return report
+
+
+def _parse_dt(value):
+    if not value:
+        return None
+    try:
+        return datetime.fromisoformat(str(value).replace('Z', '+00:00')).replace(
+            tzinfo=None)
+    except (ValueError, TypeError):
+        return None
+
+
 def export_scope_to_share(scopename, phase, shareroot):
     """Write a scope's current published JSON to the share, backing up the old
     file to _meta/history first. Returns the written path."""