"""Endpoints plugin API endpoints (read + dashboard).

Ingest is NOT here: agent reports arrive on the generic collector route
POST /api/collector/endpoints (ADR-006), authorized by a collector-scoped
service token, and are handled by EndpointsPlugin.apply_collector_payload.
These routes are the human-facing read surface, gated by user JWT + RBAC.
"""

import hashlib
import hmac
import json
import os
import time
from datetime import datetime, timedelta, timezone

from flask import Blueprint, request, current_app
from flask_jwt_extended import current_user

from shopdb.api import (
    db, Asset, Setting, success_response, error_response, paginated_response,
    ErrorCodes, get_pagination_params, paginate_query, require_permission,
    require_role, caller_holds, caller_is_role, mint_service_token,
    list_service_tokens,
    revoke_service_tokens,
    AuditLog, audit_log, collector_authorized, User,
    Application, AppVersion, Script,
)

from ..models import (
    EndpointState, EndpointEvent, EndpointCommand, EndpointApplication,
    EndpointHealthState, OperatorReport,
)
from ..commands import COMMAND_ALLOWLIST, is_allowed_action, ALLOWED_ACTIONS_FOR, row_actions_for, requires_target, action_targetkind, elevated_permission_for, PLACE_POLICY_HOLD, ACTIVE_POLICY_HOLDS, HOLDABLE_ACTIONS_FOR, RELEASE_POLICY_HOLD, ACTIVE_LOCK_FOR, PLACE_ENDPOINT_LOCK, RELEASE_ENDPOINT_LOCK
from ..eventcatalog import CRITICAL_EVENT_IDS, event_severity, event_title

endpoints_bp = Blueprint('endpoints', __name__)


def _naive_utcnow():
    return datetime.now(timezone.utc).replace(tzinfo=None)


# What counts as RUNNING is defined once, by the reconciler, and imported here.
# The panel and the reconciler read the same agent snapshot, so a disagreement
# would paint a service green and flag it as drift on the same page. Importing
# api -> services is the direction the plugin layout already runs in; the
# reconciler's own import of this module is deferred, so there is no cycle.
from ..services.reconciler import RUNNING_STATES as _RUNNING_STATES


def _service_running(entry):
    raw = (entry.get('state') or entry.get('status') or '').strip().lower()
    return raw in _RUNNING_STATES


# Paused, plus the four transitional states Windows reports while a service is
# on its way somewhere.
_TRANSITIONAL_STATES = ('paused', 'start pending', 'stop pending',
                        'continue pending', 'pause pending', 'starting',
                        'stopping', 'pausing')


def _service_transitional(raw):
    return (raw or '').strip().lower() in _TRANSITIONAL_STATES


def _service_name(entry):
    return entry.get('name') or entry.get('service') or 'Unknown'


def _format_uptime(seconds):
    """Whole-seconds uptime -> 'Nd Nh Nm', or None."""
    if seconds is None:
        return None
    try:
        seconds = int(seconds)
    except (TypeError, ValueError):
        return None
    days, rem = divmod(seconds, 86400)
    hours, rem = divmod(rem, 3600)
    minutes = rem // 60
    parts = []
    if days:
        parts.append('{}d'.format(days))
    if hours:
        parts.append('{}h'.format(hours))
    parts.append('{}m'.format(minutes))
    return ' '.join(parts)


def _state_json(state, attr):
    """Parse one of the EndpointState inventory JSON blobs.

    These columns are written straight from the agent snapshot, so a partial
    or malformed blob from one bad check-in must not take the whole panel
    down. Anything unparseable reads as absent.
    """
    raw = getattr(state, attr, None) if state else None
    if not raw:
        return None
    try:
        return json.loads(raw)
    except (TypeError, ValueError):
        return None


def _format_bytes(value):
    """Byte count -> 'N.N GB'. Sizes here are disks and RAM, never sub-MB."""
    if value is None:
        return None
    try:
        value = float(value)
    except (TypeError, ValueError):
        return None
    for unit in ('B', 'KB', 'MB', 'GB', 'TB'):
        if abs(value) < 1024.0 or unit == 'TB':
            return '{:.1f} {}'.format(value, unit) if unit != 'B' else '{:.0f} B'.format(value)
        value /= 1024.0
    return None


def _format_agent_date(value):
    """Agent timestamps are ISO strings, not datetimes; show the date only.

    A BIOS release date or an OS install date is only ever read at day
    resolution, and the agent sends a full 7-digit-fraction UTC stamp.
    """
    if not value:
        return None
    text = str(value)
    return text[:10] if len(text) >= 10 else text


# A Windows event message runs to several hundred words. It is shown in an
# expander that is closed by default, so the cap here only exists to keep one
# pathological message from bloating the payload - not to fit a line. Blank
# runs are collapsed but real line breaks are kept, because the body is read
# as a block once opened.
_EVENT_BODY_LIMIT = 4000


def _event_body(event):
    """Readable message body for one curated event, bounded and normalized.

    Windows messages arrive padded with long runs of blank lines that render
    as ragged gaps. Runs of blank lines collapse to one and trailing spaces
    go, but single newlines survive so the text keeps its shape when opened.
    """
    text = (getattr(event, 'eventmessage', None) or '').strip()
    if not text:
        return None
    lines = [line.rstrip() for line in text.splitlines()]
    cleaned = []
    for line in lines:
        if not line and cleaned and not cleaned[-1]:
            continue
        cleaned.append(line)
    text = '\n'.join(cleaned).strip()
    if not text:
        return None
    if len(text) <= _EVENT_BODY_LIMIT:
        return text
    return text[:_EVENT_BODY_LIMIT].rstrip() + '...'


# =============================================================================
# Endpoint state
# =============================================================================

@endpoints_bp.route('', methods=['GET'], strict_slashes=False)
@endpoints_bp.route('/', methods=['GET'], strict_slashes=False)
@require_permission('endpoints.fleet')
def list_endpoints():
    """List reported endpoint states.

    Registered as '' so the rule is '/api/endpoints', with no trailing slash.
    CONTRIBUTING.md's API section spells endpoints that way ('/api/machines'),
    and every other plugin's list route follows it - this was the only route in
    331 paths that did not.

    It was not cosmetic. A rule ending in a slash makes Flask answer the
    unslashed URL with a 308 to an ABSOLUTE url. On dev that resolves and
    nobody notices. Behind the IIS subpath it does not: with no ProxyFix,
    Werkzeug builds that Location from the scheme and host it believes it has,
    so it emitted http:// and dropped the /shopdb prefix. The browser refuses
    to follow an https page to http, the list request failed, and the page
    rendered "No endpoints found" - indistinguishable from a fleet with nothing
    in it. The agent had been reporting correctly the whole time.

    Joins through Computer so each row carries computerid and can deep-link
    into the existing PC detail page (/pcs/{computerid}). Endpoint state only
    exists for hostnames resolved to a Computer, so the join is not lossy.

    Sorted WORST FIRST by default: never-reported, then quiet, then reporting.
    It used to sort by lastreportedat descending, which floated the healthiest
    endpoints to the top and buried the ones somebody needs to look at.

    Each row also carries what the fleet view is actually for: which MACHINE
    depends on this PC, its PC type, and a health rollup. On a shop floor a PC
    matters because a machine stops without it, so that is the column an
    operator reads first.
    """
    from plugins.computers.models import Computer, ComputerType

    page, per_page = get_pagination_params(request)
    cutoff = _naive_utcnow() - timedelta(hours=_quiet_hours())

    # Status as a sortable expression rather than a post-hoc label, so
    # worst-first survives pagination. 0 never reported, 1 quiet, 2 reporting.
    statusrank = db.case(
        (EndpointState.lastreportedat.is_(None), 0),
        (EndpointState.lastreportedat < cutoff, 1),
        else_=2,
    )

    query = (db.session.query(EndpointState, Computer.computerid,
                              Computer.computertypeid, ComputerType.computertype,
                              Asset.assetnumber, Asset.name,
                              statusrank.label('statusrank'))
             .join(Asset, Asset.assetid == EndpointState.assetid)
             .join(Computer, Computer.assetid == EndpointState.assetid)
             .outerjoin(ComputerType,
                        ComputerType.computertypeid == Computer.computertypeid)
             .filter(Asset.isactive.is_(True)))

    # Search covers hostname, asset number and the machine number a PC serves.
    # Hostname alone was too narrow: people look a PC up by the bay it runs.
    if search := (request.args.get('search') or '').strip():
        like = f'%{search}%'
        machinehits = _assetids_serving_machine_like(like)
        clauses = [EndpointState.hostname.ilike(like), Asset.assetnumber.ilike(like),
                   Asset.name.ilike(like)]
        if machinehits:
            clauses.append(EndpointState.assetid.in_(machinehits))
        query = query.filter(db.or_(*clauses))

    if status := request.args.get('status'):
        wanted = {'never': 0, 'quiet': 1, 'reporting': 2}.get(status.lower())
        if wanted is not None:
            query = query.filter(statusrank == wanted)

    if computertypeid := request.args.get('computertypeid'):
        query = query.filter(Computer.computertypeid == computertypeid)
    if locationid := request.args.get('locationid'):
        query = query.filter(Asset.locationid == locationid)
    if businessunitid := request.args.get('businessunitid'):
        query = query.filter(Asset.businessunitid == businessunitid)

    sort = (request.args.get('sort') or 'status').lower()
    sorts = {
        'status':       (statusrank.asc(), EndpointState.hostname.asc()),
        'hostname':     (EndpointState.hostname.asc(),),
        'lastreported': (EndpointState.lastreportedat.desc(),),
        'uptime':       (EndpointState.uptimeseconds.desc(),),
    }
    query = query.order_by(*sorts.get(sort, sorts['status']))

    items, total = paginate_query(query, page, per_page)

    # Batch the per-row extras for this PAGE only. Doing them inline would be
    # one query per endpoint for machines and another for events.
    assetids = [state.assetid for state, *_ in items]
    machines = _machines_for_assets(assetids)
    states = {state.assetid: state for state, *_ in items}
    computertypes = {state.assetid: computertypeid
                     for state, _computerid, computertypeid, *_ in items}
    health = _health_for_assets(states, computertypes)

    data = []
    for (state, computerid, computertypeid, computertype, assetnumber, alias,
         rank) in items:
        row = state.to_dict()
        row['computerid'] = computerid
        row['assetnumber'] = assetnumber
        # Named 'alias' rather than 'name' because the fleet row already
        # carries a hostname; 'name' next to 'hostname' reads as a duplicate.
        row['alias'] = alias
        row['computertypeid'] = computertypeid
        row['computertype'] = computertype
        row['uptimedisplay'] = _format_uptime(state.uptimeseconds)
        row['status'] = ('never', 'quiet', 'reporting')[rank]
        row['machines'] = machines.get(state.assetid, [])
        row['health'] = health.get(state.assetid)
        data.append(row)

    return paginated_response(data, page, per_page, total)


def _assetids_serving_machine_like(like):
    """Asset ids of PCs whose controlled machine matches a search term.

    Checked in BOTH directions on purpose. The canonical edge is PC controls
    machine, but production also holds reciprocal rows, so matching only
    source->target silently drops half the fleet from a machine-number search.
    """
    from shopdb.api import AssetRelationship, RelationshipType, AssetType

    controls = (RelationshipType.query
                .filter(db.func.lower(RelationshipType.relationshiptype) == 'controls')
                .first())
    if controls is None:
        return []

    # Only MACHINE assets. A PC also 'controls' printers and other kit, and
    # matching any controlled asset put office locations in the machine column.
    other = db.aliased(Asset)
    othertype = db.aliased(AssetType)
    forward = (db.session.query(AssetRelationship.sourceassetid)
               .join(other, other.assetid == AssetRelationship.targetassetid)
               .join(othertype, othertype.assettypeid == other.assettypeid)
               .filter(AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
                       db.func.lower(othertype.assettype) == 'machine',
                       other.assetnumber.ilike(like)))
    backward = (db.session.query(AssetRelationship.targetassetid)
                .join(other, other.assetid == AssetRelationship.sourceassetid)
                .join(othertype, othertype.assettypeid == other.assettypeid)
                .filter(AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
                        db.func.lower(othertype.assettype) == 'machine',
                        other.assetnumber.ilike(like)))
    return [r[0] for r in forward.union(backward).all()]


def _machines_for_assets(assetids):
    """{assetid: [{assetid, assetnumber, name}]} for the machines each PC serves.

    Both edge directions again, deduplicated by machine asset id. One query for
    the whole page rather than one per row.
    """
    from shopdb.api import AssetRelationship, RelationshipType, AssetType

    if not assetids:
        return {}

    controls = (RelationshipType.query
                .filter(db.func.lower(RelationshipType.relationshiptype) == 'controls')
                .first())
    if controls is None:
        return {}

    other = db.aliased(Asset)
    othertype = db.aliased(AssetType)
    forward = (db.session.query(AssetRelationship.sourceassetid.label('pc'),
                                other.assetid, other.assetnumber, other.name)
               .join(other, other.assetid == AssetRelationship.targetassetid)
               .join(othertype, othertype.assettypeid == other.assettypeid)
               .filter(AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
                       db.func.lower(othertype.assettype) == 'machine',
                       AssetRelationship.sourceassetid.in_(assetids)))
    backward = (db.session.query(AssetRelationship.targetassetid.label('pc'),
                                 other.assetid, other.assetnumber, other.name)
                .join(other, other.assetid == AssetRelationship.sourceassetid)
                .join(othertype, othertype.assettypeid == other.assettypeid)
                .filter(AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
                        db.func.lower(othertype.assettype) == 'machine',
                        AssetRelationship.targetassetid.in_(assetids)))

    out = {}
    for pc, machineid, assetnumber, name in forward.union(backward).all():
        seen = out.setdefault(pc, [])
        if not any(m['assetid'] == machineid for m in seen):
            seen.append({'assetid': machineid, 'assetnumber': assetnumber, 'name': name})
    return out


def _health_for_assets(states, computertypes):
    """{assetid: {'level', 'label'}} for a page, from the cached rollup.

    ONE query for the page instead of the old per-page critical-event
    aggregate, and no ranking work here - the verdict was decided by
    services/health.py in the same commit as the telemetry it came from.

    Endpoints with no cached row have not checked in since the rollup shipped.
    Those are evaluated read-only for display rather than left blank, because a
    blank Health column reads as "fine" and not doing that is this page's whole
    job. Nothing is persisted on a GET: writing here would invent a firstseenat
    no agent ever observed, and would make a page load look like an
    observation.
    """
    if not states:
        return {}

    from ..services.health import _verdict

    assetids = list(states)
    cached = {row.assetid: row for row in
              EndpointHealthState.query
              .filter(EndpointHealthState.assetid.in_(assetids))
              .all()}

    now = _naive_utcnow()
    out = {}
    for assetid, state in states.items():
        row = cached.get(assetid)
        if row is not None:
            out[assetid] = {'level': row.level, 'label': row.label}
            continue
        verdict = _verdict(state, computertypes.get(assetid), assetid, now)
        out[assetid] = {'level': verdict['level'], 'label': verdict['label']}
    return out


@endpoints_bp.route('/<int:asset_id>', methods=['GET'])
@require_permission('endpoints.view')
def get_endpoint(asset_id: int):
    """Get the current state for one endpoint by asset id.

    Carries the same reporting/quiet/never verdict the fleet list uses. The
    quiet window is a site setting, so the SERVER decides it: a caller that
    guessed a threshold in the browser would disagree with the fleet list on
    the same endpoint the moment a site changed the setting.
    """
    state = EndpointState.query.filter_by(assetid=asset_id).first()

    if not state:
        return error_response(
            ErrorCodes.NOT_FOUND,
            f'No endpoint state for asset {asset_id}',
            http_code=404,
        )

    cutoff = _naive_utcnow() - timedelta(hours=_quiet_hours())
    if state.lastreportedat is None:
        status = 'never'
    elif state.lastreportedat < cutoff:
        status = 'quiet'
    else:
        status = 'reporting'

    data = state.to_dict()
    data['status'] = status
    data['uptimedisplay'] = _format_uptime(state.uptimeseconds)
    return success_response(data)


@endpoints_bp.route('/<int:asset_id>/events', methods=['GET'])
@require_permission('endpoints.view')
def list_endpoint_events(asset_id: int):
    """List curated Windows events for one endpoint, newest first."""
    page, per_page = get_pagination_params(request)

    query = EndpointEvent.query.filter_by(assetid=asset_id)

    if eventid := request.args.get('eventid'):
        try:
            query = query.filter(EndpointEvent.eventid == int(eventid))
        except (TypeError, ValueError):
            return error_response(ErrorCodes.VALIDATION_ERROR,
                                  'eventid must be an integer')

    query = query.order_by(EndpointEvent.eventtime.desc())

    items, total = paginate_query(query, page, per_page)
    data = [e.to_dict() for e in items]

    return paginated_response(data, page, per_page, total)


# =============================================================================
# Asset panels (ADR-010) - shaped for the generic PluginAssetPanels renderer
# =============================================================================

def _drift_for_state(state):
    """Parse the cached warn-only reconciler drift list off an EndpointState.

    driftjson is written on every check-in by the endpoints ingest: the monitors
    site policy expects running that the latest snapshot did not report running.
    Null means never computed; an empty list means computed with no drift.
    Tolerant of a malformed blob so a bad cache never breaks the panel.
    """
    raw = getattr(state, 'driftjson', None) if state else None
    if not raw:
        return []
    try:
        parsed = json.loads(raw)
    except (TypeError, ValueError):
        return []
    return parsed if isinstance(parsed, list) else []


def _dell_updates_summary(state):
    """Parse the cached Dell Command | Update summary off an EndpointState.

    dellupdatesjson is written on check-in from the shopdb-dellupdates agent
    utility, which reads the log the \\ShopDB-DellCommandScan task produces.
    Null means the endpoint has never reported a scan. Tolerant of a malformed
    blob so a bad cache never breaks the panel.
    """
    raw = getattr(state, 'dellupdatesjson', None) if state else None
    if not raw:
        return None
    try:
        parsed = json.loads(raw)
    except (TypeError, ValueError):
        return None
    return parsed if isinstance(parsed, dict) else None


def _dell_updates_value(state):
    """Render the informational 'Pending Dell updates' status value.

    Informational only - this row never claims an endpoint needs attention,
    because a pending optional driver is not a support problem. Three states are
    kept distinct on purpose, and all three are VISIBLE:

      never scanned -> 'Never scanned'. Deliberately not 'No': those are
                       different facts, and claiming 'No' would tell an operator
                       the machine is current when nobody has ever checked. It
                       is also not hidden - a missing row reads as a broken
                       panel, so the unknown state says so out loud.
      scanned clean -> 'No'.
      pending       -> 'Yes (n)', with the count because one optional update and
                       fourteen urgent ones warrant different urgency.
    """
    summary = _dell_updates_summary(state)
    if not summary:
        return 'Never scanned'
    status = summary.get('status')
    if status == 'error':
        # The scan log exists but could not be read. A different problem from
        # never having scanned, and worth saying so rather than implying the
        # endpoint was never checked.
        return 'Unknown (scan unreadable)'
    if status != 'ok':
        return 'Never scanned'
    try:
        count = int(summary.get('count') or 0)
    except (TypeError, ValueError):
        return 'Unknown (scan unreadable)'
    if count <= 0:
        return 'No'
    return 'Yes ({})'.format(count)


@endpoints_bp.route('/asset/<int:asset_id>/status', methods=['GET'])
@require_permission('endpoints.view')
def panel_status(asset_id: int):
    """Keyvalue panel: the endpoint's current status snapshot.

    Single indexed lookup on endpointstates.assetid. Returns the {fields:[...]}
    shape the keyvalue renderer expects; an absent state yields no fields so the
    panel hides itself.
    """
    state = EndpointState.query.filter_by(assetid=asset_id).first()
    if not state:
        return success_response({'fields': []})

    def field(label, value, mono=False, fmt=None):
        entry = {'label': label, 'value': value, 'mono': mono}
        if fmt:
            entry['format'] = fmt
        return entry

    fields = [
        field('Logged in user', state.loggedinuser),
        field('Uptime', _format_uptime(state.uptimeseconds)),
        field('Idle', _format_uptime(state.idleseconds)),
        field('Last boot',
              state.lastboottime.isoformat() + 'Z' if state.lastboottime else None,
              fmt='datetime'),
        field('CPU %', state.cpupercent),
        field('Memory %', state.memorypercent),
        field('Agent', state.agentversion, mono=True),
        field('Tray', state.trayversion, mono=True),
        field('Last reported',
              state.lastreportedat.isoformat() + 'Z' if state.lastreportedat else None,
              fmt='datetime'),
    ]
    fields = [f for f in fields if f['value'] not in (None, '')]

    # Informational only, and appended after the null-filter above because it
    # always renders - including an explicit "Never scanned", which a hidden row
    # would leave looking like a broken panel.
    fields.append(field('Pending Dell updates', _dell_updates_value(state)))

    # Warn-only reconciler summary: monitors site policy expects to be running
    # that the last check-in did not report as running. Surfaced as a status
    # field so an operator sees the drift on the panel; nothing is auto-started.
    drift = _drift_for_state(state)
    if drift:
        names = ', '.join(entry.get('displayname') or entry.get('name')
                          for entry in drift)
        fields.append(field(
            'Needs attention',
            '{} monitor(s) expected running but not reported: {}'.format(
                len(drift), names)))

    return success_response({'fields': fields})


@endpoints_bp.route('/asset/<int:asset_id>/hardware', methods=['GET'])
@require_permission('endpoints.view')
def panel_hardware(asset_id: int):
    """Keyvalue panel: processor, memory and system chassis inventory.

    Only reported by agent 1.7.0 and later. An older agent leaves these
    columns null, so the panel yields no fields and hides itself rather than
    rendering a card full of blanks.
    """
    state = EndpointState.query.filter_by(assetid=asset_id).first()
    if not state:
        return success_response({'fields': []})

    processor = _state_json(state, 'processorjson') or {}
    memory = _state_json(state, 'memoryjson') or {}

    fields = [
        {'label': 'Processor', 'value': processor.get('name')},
        {'label': 'Cores', 'value': processor.get('cores')},
        {'label': 'Logical processors', 'value': processor.get('logicalprocessors')},
        {'label': 'Max clock',
         'value': '{} MHz'.format(processor['maxclockmhz'])
                  if processor.get('maxclockmhz') else None},
        {'label': 'Total memory', 'value': _format_bytes(memory.get('totalbytes'))},
        {'label': 'Free memory', 'value': _format_bytes(memory.get('freebytes'))},
    ]
    fields = [f for f in fields if f['value'] not in (None, '')]
    return success_response({'fields': fields})


# Wire values to operator-readable text. The wire keys are stable contract;
# these labels are not, so they live here rather than in the agent.
_REBOOT_REASON_LABELS = {
    'componentbasedservicing': 'servicing',
    'windowsupdate': 'Windows Update',
    'pendingfilerename': 'files staged for replacement',
}


@endpoints_bp.route('/asset/<int:asset_id>/osinfo', methods=['GET'])
@require_permission('endpoints.view')
def panel_osinfo(asset_id: int):
    """Keyvalue panel: operating system, firmware and domain membership.

    OS and BIOS are one card on purpose: both answer "what is this box
    running", and split across two cards each would be three short rows.
    """
    state = EndpointState.query.filter_by(assetid=asset_id).first()
    if not state:
        return success_response({'fields': []})

    osinfo = _state_json(state, 'osinfojson') or {}
    bios = _state_json(state, 'biosjson') or {}
    bootinfo = _state_json(state, 'bootinfojson') or {}

    pendingreboot = None
    if state.pendingreboot is not None:
        pendingreboot = 'Yes' if state.pendingreboot else 'No'
        # Say WHICH signal fired when we know. A staged file rename means the
        # box is mid-servicing; a lone servicing key can sit set for weeks on a
        # machine nobody rebooted. Same "Yes", very different urgency.
        reasons = _state_json(state, 'rebootreasonsjson')
        if state.pendingreboot and isinstance(reasons, list) and reasons:
            pendingreboot = '{} ({})'.format(
                pendingreboot, ', '.join(_REBOOT_REASON_LABELS.get(r, r)
                                         for r in reasons if isinstance(r, str)))

    # Only worth a row when fast startup actually applies. On a box with it
    # disabled and a cold last boot there is nothing to warn about, and a
    # permanent "Cold" row is noise on a card that is already dense.
    boottype = bootinfo.get('lastboottype')
    if boottype in (None, '', 'Cold', 'Unknown'):
        boottype = None
    elif boottype == 'Fast startup':
        # Uptime is measured from the resume, not the last true boot, so it
        # understates how long the kernel has actually been up. Said plainly
        # because this is the field that makes uptime read wrong.
        boottype = 'Fast startup (uptime not since cold boot)'

    fields = [
        {'label': 'OS version', 'value': osinfo.get('version'), 'mono': True},
        {'label': 'Build', 'value': osinfo.get('build'), 'mono': True},
        {'label': 'Architecture', 'value': osinfo.get('architecture')},
        {'label': 'Installed', 'value': _format_agent_date(osinfo.get('installdate'))},
        {'label': 'Domain', 'value': state.domain},
        {'label': 'BIOS vendor', 'value': bios.get('vendor')},
        {'label': 'BIOS version', 'value': bios.get('version'), 'mono': True},
        {'label': 'BIOS date', 'value': _format_agent_date(bios.get('releasedate'))},
        {'label': 'Last boot', 'value': boottype},
        {'label': 'Pending reboot', 'value': pendingreboot},
    ]
    fields = [f for f in fields
              if f['value'] not in (None, '')]
    return success_response({'fields': fields})


def _storage_rows(disks):
    """Turn the agent's volume list into display rows.

    freepercent is computed by the agent from 1.7.0; for a snapshot written
    by an older agent it is derived here from the byte counts so the column
    is not empty for the one field an operator actually scans.
    """
    if not isinstance(disks, list):
        return []

    rows = []
    for disk in disks:
        if not isinstance(disk, dict):
            continue
        freepercent = disk.get('freepercent')
        if freepercent is None:
            size = disk.get('sizebytes') or 0
            free = disk.get('freebytes')
            if size and free is not None:
                freepercent = round(free * 100.0 / size, 1)
        rows.append({
            'drive': disk.get('drive'),
            'label': disk.get('label') or '',
            'filesystem': disk.get('filesystem'),
            'size': _format_bytes(disk.get('sizebytes')),
            'free': _format_bytes(disk.get('freebytes')),
            'freepercent': '{}%'.format(freepercent) if freepercent is not None else None,
        })
    return rows


_STORAGE_COLUMNS = [
    {'key': 'drive', 'label': 'Drive'},
    {'key': 'label', 'label': 'Label'},
    {'key': 'filesystem', 'label': 'Filesystem'},
    {'key': 'size', 'label': 'Size'},
    {'key': 'free', 'label': 'Free'},
    {'key': 'freepercent', 'label': 'Free %'},
]

_DISKHEALTH_COLUMNS = [
    {'key': 'model', 'label': 'Drive'},
    {'key': 'mediatype', 'label': 'Media'},
    {'key': 'bustype', 'label': 'Bus'},
    {'key': 'healthstatus', 'label': 'Health'},
    {'key': 'verdict', 'label': 'Verdict'},
    {'key': 'source', 'label': 'Source'},
]


@endpoints_bp.route('/asset/<int:asset_id>/storage', methods=['GET'])
@require_permission('endpoints.view')
def panel_storage(asset_id: int):
    """Tabs panel: volume capacity and physical drive health as two sections.

    One card because an operator looking at a full disk and an operator looking
    at a dying disk are usually the same person in the same minute. TWO
    SECTIONS because the two do not share a key: storage is per volume (C:, D:)
    and health is per physical device, and one drive can back several volumes
    or none the operator sees. Nothing is reconciled across the sections -
    this is not the mapped-drive merge, where two sources described the same
    drive and the worse status won.

    Both sections are emitted even when empty so the section keeps its own
    empty message; a dropped section would take its tab with it and read as
    "no disks" rather than "nothing reported".
    """
    state = EndpointState.query.filter_by(assetid=asset_id).first()
    disks = _state_json(state, 'disksjson') if state else None
    drives = _state_json(state, 'diskhealthjson') if state else None
    return success_response({'sections': [
        {
            'label': 'Volumes',
            'columns': _STORAGE_COLUMNS,
            'rows': _storage_rows(disks),
            'empty': 'No disks reported.',
        },
        {
            'label': 'Disk health',
            'columns': _DISKHEALTH_COLUMNS,
            'rows': _diskhealth_rows(drives),
            'empty': 'No disk health reported.',
        },
    ]})


def _diskhealth_rows(drives):
    """Turn the agent's disk health list into display rows, worst first.

    Both agent sources are kept rather than merged. MSFT_PhysicalDisk is the
    only source that reports NVMe at all, and Win32_DiskDrive covers hardware
    and hosts where the Storage namespace is unavailable. When both answer for
    the same hardware, two rows that disagree is the useful signal; collapsing
    them would hide exactly the case worth looking at.
    """
    if not isinstance(drives, list):
        return []

    rows = []
    for drive in drives:
        if not isinstance(drive, dict):
            continue
        predictfailure = drive.get('predictfailure')
        if predictfailure is None:
            verdict = 'Unknown'
        elif predictfailure:
            verdict = 'Failure predicted'
        else:
            verdict = 'OK'
        rows.append({
            'deviceid': drive.get('deviceid'),
            'model': drive.get('model') or '',
            'mediatype': drive.get('mediatype') or '',
            'bustype': drive.get('bustype') or '',
            'healthstatus': drive.get('healthstatus') or 'Unknown',
            'verdict': verdict,
            'source': drive.get('source') or '',
        })

    # Failure first, then unknown, then OK. Unknown outranks OK because a drive
    # that stopped answering is a weaker claim of health than one that said OK.
    # An unrecognised verdict sorts as unknown rather than last, so a future
    # verdict string cannot bury itself under the healthy drives.
    order = {'Failure predicted': 0, 'Unknown': 1, 'OK': 2}
    rows.sort(key=lambda row: (order.get(row['verdict'], 1),
                               row['deviceid'] or ''))
    return rows


@endpoints_bp.route('/asset/<int:asset_id>/diskhealth', methods=['GET'])
@require_permission('endpoints.view')
def panel_diskhealth(asset_id: int):
    """Physical drive health rows.

    No longer backs a panel of its own - the Storage card renders it as a
    section. Kept as a route because it is a stable read for callers outside
    the panel renderer and removing it would break them for a UI-only change.

    Rows are ordered so anything predicting failure sorts first: a failing
    drive must not be below the fold on a machine with several disks.
    """
    state = EndpointState.query.filter_by(assetid=asset_id).first()
    drives = _state_json(state, 'diskhealthjson') if state else None
    return success_response(_diskhealth_rows(drives))


@endpoints_bp.route('/asset/<int:asset_id>/listeners', methods=['GET'])
@require_permission('endpoints.view')
def panel_listeners(asset_id: int):
    """Table panel: TCP sockets in LISTEN, with the owning process.

    The listening set is the box's exposed surface, so it is sorted by port
    to make an unexpected one easy to spot. The full connection table is a
    deliberately on-demand command, not standard telemetry.
    """
    state = EndpointState.query.filter_by(assetid=asset_id).first()
    ports = _state_json(state, 'portsjson') if state else None
    if not isinstance(ports, list):
        return success_response([])

    rows = []
    for entry in ports:
        if not isinstance(entry, dict):
            continue
        # Sample-data rows predate the agent listener collector and carry a
        # flat port/process shape; read both rather than showing them blank.
        port = entry.get('localport', entry.get('port'))
        rows.append({
            'localport': port,
            'localaddress': entry.get('localaddress'),
            'protocol': (entry.get('protocol') or 'tcp').upper(),
            'processname': entry.get('processname') or entry.get('process'),
            'processid': entry.get('processid'),
        })
    rows.sort(key=lambda r: (r['localport'] is None, r['localport']))
    return success_response(rows)


def _drive_letter(value):
    """Normalise a drive to its bare letter so the two sources can be paired.

    net use reports 'Z:' and Get-SmbMapping reports 'Z:' or occasionally a
    trailing separator, so compare on the letter alone.
    """
    text = (value or '').strip().rstrip('\\/')
    return text.upper() or None


# Worst-first. The panel shows the worse of the two sources for a drive, so a
# share that one source calls OK and the other calls Unavailable is reported as
# Unavailable: a drive that is broken in either view is broken for the operator.
#
# WARNING - ranks must stay DISTINCT across statuses that can genuinely differ.
# _worse_drive_status resolves by min(), which breaks a tie by argument order,
# so two statuses sharing a rank make the verdict depend on which SOURCE
# happened to report which - the same drive reads healthy or broken based on
# enumeration order. Deliberate synonyms are fine (disconnected/unavailable,
# ok/connected) because either answer is the same answer. Adding a NEW status at
# an existing rank is not. test_which_source_reported_a_status_does_not_change
# _the_verdict in tests/test_plugins/test_endpoints_mapped_drives.py asserts
# both directions of each pair and is what catches this; the single-direction
# tests stay green through it.
_DRIVE_STATUS_RANK = {
    'unavailable': 0,
    'disconnected': 0,
    'paused': 1,
    'reconnecting': 2,
    'ok': 3,
    'connected': 3,
}


def _worse_drive_status(left, right):
    """Return whichever status is worse, preferring a known one over None.

    An unrecognised status ranks as Reconnecting, deliberately below OK: a
    status we cannot interpret must not be assumed healthy.
    """
    candidates = [s for s in (left, right) if s]
    if not candidates:
        return None
    unknown_rank = _DRIVE_STATUS_RANK['reconnecting']
    return min(
        candidates,
        key=lambda s: _DRIVE_STATUS_RANK.get(str(s).strip().lower(), unknown_rank)
    )


def _mapped_drive_rows(context):
    """Merge the two mapped-drive views into one structured row per letter.

    Windows exposes mapped drives through Win32_NetworkConnection (the 'net use'
    table) and through Get-SmbMapping, and they do not always agree - a share can
    be Unavailable in one and OK or absent in the other. The probe collects both
    without reconciling them, so the disagreement survives to here.

    A drive present in only one source is still a drive: it is reported with the
    status that source gave it, rather than dropped for lack of a second opinion.

    WARNING - three behaviours here are load-bearing and easy to break while
    "simplifying". They are covered by tests/test_plugins/test_endpoints_mapped
    _drives.py; if you touch this, read those first:

      1. Worse-status-wins. Never report the healthier of the two sources. A
         drive that either source calls Unavailable IS unavailable to the
         operator, and rendering it OK sends someone to a share that is down.
      2. Single-source drives are kept, not dropped. Get-SmbMapping is absent on
         some builds and session types; requiring agreement would silently hide
         every mapped drive on those machines.
      3. Unknown statuses rank as Reconnecting, not as OK. An unrecognised
         string must never be optimistically treated as healthy.

    Changing any of these makes a broken drive look fine, which is a silent
    failure - the card stays green and nobody is told.
    """
    drives = {}

    netuse = context.get('netuse')
    if isinstance(netuse, list):
        for entry in netuse:
            if not isinstance(entry, dict):
                continue
            letter = _drive_letter(entry.get('localname') or entry.get('local'))
            if not letter:
                continue
            drives[letter] = {
                'remote': entry.get('remotename') or entry.get('remote') or '',
                'status': entry.get('status'),
            }

    smbmapping = context.get('smbmapping')
    if isinstance(smbmapping, list):
        for entry in smbmapping:
            if not isinstance(entry, dict):
                continue
            letter = _drive_letter(entry.get('localpath'))
            if not letter:
                continue
            existing = drives.get(letter)
            if existing:
                existing['status'] = _worse_drive_status(
                    existing.get('status'), entry.get('status'))
                existing['remote'] = existing['remote'] or entry.get('remotepath') or ''
            else:
                drives[letter] = {
                    'remote': entry.get('remotepath') or '',
                    'status': entry.get('status'),
                }

    rows = []
    for letter in sorted(drives):
        drive = drives[letter]
        # A drive neither source gave a status for is Unknown, never blank. An
        # unannotated row reads as healthy, so a failed check would render as a
        # working drive - the one outcome this card must never produce.
        status = (drive.get('status') or '').strip() or 'Unknown'
        healthy = status.lower() in ('ok', 'connected')
        rows.append({
            'drive': letter,
            'remote': (drive.get('remote') or '').strip(),
            'statuslabel': status,
            # Only unhealthy drives are coloured. Green on every working drive
            # makes the broken one just another badge in a row of badges; the
            # card's job is to make the exception findable at a glance.
            'statecolor': 'green' if healthy else 'red',
        })
    return rows


def _mapped_drives(context):
    """The mapped drives as one-line labels, for the user context card.

    Built on _mapped_drive_rows so the summary line and the card can never
    disagree about a drive's status.
    """
    labels = []
    for row in _mapped_drive_rows(context):
        label = '{} {}'.format(row['drive'], row['remote']).strip()
        # Only the problem states are annotated. Tagging every healthy drive
        # with "(OK)" would bury the one that is not.
        if row['statecolor'] != 'green':
            label = '{} ({})'.format(label, row['statuslabel'])
        labels.append(label)
    return labels


def _mapped_drives_unhealthy_first(rows):
    """Order drive rows so anything not healthy sorts first.

    A user with a dozen mappings should not have to hunt for the broken one.
    Sorting only on health keeps it stable: drive-letter order survives within
    each group, so the list does not reshuffle between refreshes.
    """
    return sorted(rows, key=lambda row: row['statecolor'] == 'green')


@endpoints_bp.route('/asset/<int:asset_id>/mappeddrives', methods=['GET'])
@require_permission('endpoints.view')
def panel_mappeddrives(asset_id: int):
    """List panel: mapped network drives with a health badge each.

    Session-scoped: mapped drives do not exist in the agent's session 0, so
    these come from the interactive user probe. Nothing reported when nobody
    is signed in, which is a valid state and not a failure.
    """
    state = EndpointState.query.filter_by(assetid=asset_id).first()
    context = _state_json(state, 'usercontextjson') if state else None
    if not isinstance(context, dict) or not context:
        return success_response([])
    return success_response(_mapped_drives_unhealthy_first(_mapped_drive_rows(context)))


@endpoints_bp.route('/asset/<int:asset_id>/usercontext', methods=['GET'])
@require_permission('endpoints.view')
def panel_usercontext(asset_id: int):
    """Keyvalue panel: session-scoped data from the interactive user probe.

    Mapped drives and user-owned listeners do not exist in the agent's
    session 0, so they are collected by a probe run in the logged-in user's
    session. Nothing is reported when no one is signed in, which is a valid
    state and not a failure.

    Mapped drives arrive from two sources that can disagree; the worse of the
    two statuses is shown so a half-broken share is never rendered as healthy.
    """
    state = EndpointState.query.filter_by(assetid=asset_id).first()
    if not state:
        return success_response({'fields': []})

    context = _state_json(state, 'usercontextjson')
    if not isinstance(context, dict) or not context:
        return success_response({'fields': []})

    drives = _mapped_drives(context)

    listeners = context.get('userlisteners')
    listenercount = len(listeners) if isinstance(listeners, list) else None

    fields = [
        {'label': 'Probed user', 'value': context.get('username')},
        {'label': 'Collected', 'value': context.get('collectedat')},
        {'label': 'Idle', 'value': _format_uptime(context.get('idleseconds'))},
        {'label': 'Mapped drives',
         'value': ', '.join(drives) if drives else None, 'mono': True},
        {'label': 'User listeners', 'value': listenercount},
    ]
    fields = [f for f in fields if f['value'] not in (None, '')]
    return success_response({'fields': fields})


@endpoints_bp.route('/asset/<int:asset_id>/telemetrymeta', methods=['GET'])
@require_permission('endpoints.view')
def panel_telemetry_meta(asset_id: int):
    """Lightweight freshness probe for the panels.

    Returns just the endpoint's lastreportedat so the WebUI can tell whether a
    genuinely new telemetry snapshot has landed after it forces a collect. The
    UI captures this before enqueuing collect_now (or a row Start/Stop), then
    polls until the stamp advances - a command reaching 'succeeded' only means
    the agent ran it, not that it has yet posted the fresh services snapshot, so
    gating the refresh on this timestamp is what avoids re-rendering stale rows.
    """
    state = EndpointState.query.filter_by(assetid=asset_id).first()
    stamp = (state.lastreportedat.isoformat() + 'Z'
             if state and state.lastreportedat else None)
    return success_response({'lastreportedat': stamp})


def _pctype_for_asset(asset_id):
    """Resolve an asset's PC type name, or None.

    Endpoints stays optional-on-computers: a missing plugin, a non-computer
    asset, or an unset type all resolve to None so the caller falls back to
    reported-only behaviour.
    """
    try:
        from plugins.computers.models import Computer, ComputerType
    except Exception:
        return None
    row = (db.session.query(ComputerType.computertype)
           .join(Computer,
                 Computer.computertypeid == ComputerType.computertypeid)
           .filter(Computer.assetid == asset_id)
           .first())
    return row[0] if row else None


def _resolved_monitors_for_asset(asset_id):
    """Monitor definitions attached to this asset's PC type by policy/shipped.

    Reuses the endpoints monitor resolver (shipped defaults plus policy-attached
    monitors). Returns a list of EndpointMonitor rows, or an empty list when the
    PC type is unknown or nothing resolves.
    """
    pctype = _pctype_for_asset(asset_id)
    if not pctype:
        return []
    try:
        from .monitors import _resolve_pctype_monitors
        return _resolve_pctype_monitors(pctype) or []
    except Exception:
        return []


def _row_action_dict(app, kind, name, running):
    """Build the inline Start/Stop action map for a services/processes row."""
    targets = row_actions_for(app, kind, name)
    if not targets:
        return None
    return {
        'running': running,
        'start': targets['start'],
        'stop': targets['stop'],
        'status': targets['status'],
        'target': targets['target'],
        'targetkind': targets['targetkind'],
    }


@endpoints_bp.route('/asset/<int:asset_id>/services', methods=['GET'])
@require_permission('endpoints.view')
def panel_services(asset_id: int):
    """List panel: monitored services/processes with a run-state badge per row.

    Reshapes the latest servicesjson snapshot into rows the list renderer maps
    to name + status badge, then MERGES the monitors that site policy (and the
    shipped defaults) attach to this asset's PC type. A policy-attached monitor
    the agent has not yet reported surfaces as an Unknown row with a Start
    control, so an operator sees the intent immediately rather than waiting for
    the next agent snapshot. Tolerant of a missing snapshot or plugin: an
    unknown state renders as an Unknown badge rather than failing.
    """
    state = EndpointState.query.filter_by(assetid=asset_id).first()
    maycommand = caller_holds('endpoints.command')

    services = []
    if state and state.servicesjson:
        try:
            services = json.loads(state.servicesjson) or []
        except (TypeError, ValueError):
            services = []

    rows = []
    # Reported identities, keyed (kind, lowercased name), so a policy monitor
    # the agent already reports is not duplicated as a synthetic row.
    reported = set()
    # Warn-only reconciler drift, keyed (kind, lowercased name), so a row that is
    # expected running but not reported running can be flagged for the operator.
    # kind may be None on an older snapshot, so match on name too.
    drift_keys = set()
    for entry in _drift_for_state(state):
        name = (entry.get('name') or '').strip().lower()
        if not name:
            continue
        drift_keys.add(((entry.get('kind') or '').lower() or None, name))
        drift_keys.add((None, name))

    def _is_drift(kind, name):
        key = (name or '').strip().lower()
        return (kind, key) in drift_keys or (None, key) in drift_keys

    for entry in services:
        if isinstance(entry, dict):
            name = entry.get('name') or entry.get('service') or 'Unknown'
            raw = (entry.get('state') or entry.get('status') or '').strip().lower()
            app = entry.get('app')
            kind = entry.get('kind')
        else:
            name, raw, app, kind = str(entry), '', None, None
        running = _service_running(entry)
        normkind = (kind or '').lower() or None
        row = {
            'name': name,
            'kind': normkind,
            'statelabel': 'Running' if running else (raw.title() or 'Unknown'),
            # Amber for paused and the four Windows transitional states. A
            # service stuck in Start Pending is failing to start on a retry
            # cycle - which is what raises event 7011 - and a paused one was
            # paused deliberately. Painting either the same red as Stopped
            # loses the distinction that tells you which to chase.
            'statecolor': ('green' if running
                           else 'yellow' if _service_transitional(raw)
                           else 'red' if raw else 'gray'),
        }
        # Only offer Start/Stop to a caller who could actually run it. The
        # emission never consulted the caller, so a read-only viewer was shown
        # controls on a shop-floor PC that 403 on click - wrong signal, and it
        # gets worse the moment guests can read this panel.
        action = _row_action_dict(app, kind, name, running) if maycommand else None
        if action:
            row['action'] = action
        if not running and _is_drift(normkind, name):
            row['drift'] = True
        rows.append(row)
        reported.add((normkind, name.strip().lower()))

    # Merge policy/shipped monitors this PC type should watch. A monitor already
    # reported by the agent stays as its live row; one not yet seen is appended
    # as an Unknown row so the operator sees the intended coverage and can Start
    # it. The monitor's displayname is the operator label (e.g. 'msedge.exe'
    # -> 'Edge'); its process/service name is the wire target for actions.
    for monitor in _resolved_monitors_for_asset(asset_id):
        kind = (monitor.kind or '').lower() or None
        osname = (monitor.processname if kind == 'process'
                  else monitor.servicename)
        osname = (osname or '').strip()
        if not osname:
            continue
        if (kind, osname.lower()) in reported:
            continue
        row = {
            'name': monitor.displayname or osname,
            'kind': kind,
            'statelabel': 'Unknown',
            'statecolor': 'gray',
        }
        action = _row_action_dict(None, kind, osname, False) if maycommand else None
        if action:
            row['action'] = action
        if _is_drift(kind, osname):
            row['drift'] = True
        rows.append(row)
        reported.add((kind, osname.lower()))

    return success_response(rows)


@endpoints_bp.route('/asset/<int:asset_id>/recentevents', methods=['GET'])
@require_permission('endpoints.view')
def panel_recent_events(asset_id: int):
    """List panel: the endpoint's most recent curated events, newest first.

    Indexed range scan on (assetid, eventtime), capped small - this is the
    at-a-glance history card, not the full report (that is the events route
    with pagination).
    """
    events = (EndpointEvent.query
              .filter_by(assetid=asset_id)
              .order_by(EndpointEvent.eventtime.desc())
              .limit(10)
              .all())

    rows = [{
        'title': event_title(e.eventid),
        'severity': event_severity(e.eventid),
        'eventtime': e.eventtime.isoformat() + 'Z' if e.eventtime else None,
    } for e in events]
    return success_response(rows)


@endpoints_bp.route('/asset/<int:asset_id>/activity', methods=['GET'])
@require_permission('endpoints.view')
def panel_activity(asset_id: int):
    """List panel: one chronological feed of what happened to this endpoint.

    Curated Windows events and operator commands merged and sorted newest
    first. They answer the same question - what happened to this box - and
    reading them as two separate cards means interleaving them by eye.

    Both sources are already keyed by assetid, so this is two indexed reads and
    a merge sort, not a join. The core AuditLog is deliberately NOT a third
    source: its rows carry the action as the entity name and the asset only
    inside a details string, so pulling it in would mean parsing prose.

    The requester's name is withheld from callers who cannot themselves issue
    commands. A read-only viewer - a guest, on a site that grants this panel -
    has no need for the roster of who restarted what, and the feed is just as
    useful without it.
    """
    naming = caller_holds('endpoints.command')

    events = (EndpointEvent.query
              .filter_by(assetid=asset_id)
              .order_by(EndpointEvent.eventtime.desc())
              .limit(15)
              .all())

    # The newest 15 of ANY severity can be entirely routine logons while the
    # health rollup is shouting about a burst of crashes underneath them - the
    # page then contradicts itself for no visible reason. Pull the newest
    # criticals separately and merge, so what health is reporting is always
    # present in the feed that is supposed to explain it.
    criticals = (EndpointEvent.query
                 .filter(EndpointEvent.assetid == asset_id,
                         EndpointEvent.eventid.in_(CRITICAL_EVENT_IDS))
                 .order_by(EndpointEvent.eventtime.desc())
                 .limit(5)
                 .all())
    seen = {e.endpointeventid for e in events}
    events.extend(e for e in criticals if e.endpointeventid not in seen)

    commands = (EndpointCommand.query
                .filter_by(assetid=asset_id)
                .order_by(EndpointCommand.createdat.desc())
                .limit(15)
                .all())

    # requestedby holds a user ID, not a name. One batched lookup rather than a
    # query per row, and only when the caller is allowed to see the names at all.
    names = {}
    if naming:
        userids = {c.requestedby for c in commands if c.requestedby}
        if userids:
            names = {u.userid: (u.username or '')
                     for u in User.query.filter(User.userid.in_(userids)).all()}

    rows = []
    for event in events:
        # The provider name is what distinguishes two events sharing an ID,
        # and the message body is the line an operator actually reads. Both
        # are already stored; the feed just was not showing them.
        source = event.eventsource or event.eventchannel
        detail = 'Windows event {}'.format(event.eventid)
        if source:
            detail = '{} - {}'.format(detail, source)
        rows.append({
            'title': event_title(event.eventid),
            'severity': event_severity(event.eventid),
            'detail': detail,
            'body': _event_body(event),
            'eventtime': event.eventtime,
        })

    for command in commands:
        label = COMMAND_ALLOWLIST.get(command.action, {}).get('label') or command.action
        target = ' on {}'.format(command.target) if command.target else ''
        who = ''
        if naming and command.requestedby:
            username = names.get(command.requestedby)
            if username:
                who = ' by {}'.format(username)
        rows.append({
            'title': '{}{}'.format(label, target),
            # A failed command is the one an operator needs to notice.
            'severity': 'critical' if command.status == EndpointCommand.STATUS_FAILED else 'info',
            'detail': '{}{}{}'.format('Command - ', (command.status or 'queued').title(), who),
            # A failed command's reason is the whole point of noticing it.
            'body': (command.resultmessage or '').strip() or None,
            'eventtime': command.createdat,
        })

    # Undated rows sort last rather than crashing the comparison.
    rows.sort(key=lambda r: (r['eventtime'] is not None, r['eventtime']), reverse=True)

    # Trim to the cap without dropping the criticals that were fetched on
    # purpose: they keep their place in time, but they survive the cut.
    if len(rows) > 15:
        keep = [r for r in rows if r['severity'] == 'critical'][:5]
        rest = [r for r in rows if r not in keep]
        rows = sorted(keep + rest[:15 - len(keep)],
                      key=lambda r: (r['eventtime'] is not None, r['eventtime']),
                      reverse=True)

    # The list renderer reads a badge colour from a ROW FIELD, so without this
    # every badge falls back to grey and the feed loses the one signal it is
    # there to carry.
    colours = {'critical': 'red', 'warning': 'yellow', 'info': 'gray'}

    return success_response([{
        'title': r['title'],
        'severity': r['severity'],
        'severitycolor': colours.get(r['severity'], 'gray'),
        'detail': r['detail'],
        'body': r['body'],
        'eventtime': r['eventtime'].isoformat() + 'Z' if r['eventtime'] else None,
    } for r in rows[:15]])


def _link_label(speedbps, fullduplex):
    """Render NIC link speed and duplex as one cell.

    Null speed means the agent could not read link state OR the adapter is
    disconnected - verified on a live box that a disconnected NIC returns empty
    values. Either way we do not know, so it reads '-' rather than 0 Mb.

    Half duplex is called out explicitly because that is the actionable case: a
    NIC that negotiated half duplex on a switch port is a cable or port fault
    that otherwise presents as "the machine is slow sometimes". Full duplex is
    the normal case and gets no annotation, so the only thing standing out in
    the column is the thing worth looking at.
    """
    if not speedbps:
        return '-'

    if speedbps >= 1000000000:
        speed = '{:g} Gb'.format(speedbps / 1000000000)
    else:
        speed = '{:g} Mb'.format(speedbps / 1000000)

    if fullduplex is False:
        return '{} half duplex'.format(speed)
    return speed


@endpoints_bp.route('/asset/<int:asset_id>/interfaces', methods=['GET'])
@require_permission('endpoints.view')
def panel_interfaces(asset_id: int):
    """Table panel: every hardware NIC the agent last reported.

    Reshapes the latest interfacesjson snapshot into one row per adapter with
    the {alias, ipv4, mac, gateway, subnet, dhcp} shape the collector emits. A
    gatewayless adapter is flagged as the likely machine-IPS link (see
    DESIGN-DECISIONS-agent.md 9.3): those NICs speak to the machine, not the
    corporate LAN. Tolerant of a missing or unexpected snapshot: an unreadable
    payload yields no rows so the panel hides itself.
    """
    state = EndpointState.query.filter_by(assetid=asset_id).first()
    if not state or not state.interfacesjson:
        return success_response([])

    try:
        interfaces = json.loads(state.interfacesjson)
    except (TypeError, ValueError):
        return success_response([])

    rows = []
    for entry in (interfaces or []):
        if not isinstance(entry, dict):
            continue
        gateway = (entry.get('gateway') or '').strip()
        ipv4 = (entry.get('ipv4') or '').strip()
        rows.append({
            'alias': entry.get('alias') or 'Unknown',
            'ipv4': ipv4 or '-',
            'mac': (entry.get('mac') or '').strip() or '-',
            'gateway': gateway or '-',
            'subnet': (entry.get('subnet') or '').strip() or '-',
            'dhcp': 'DHCP' if entry.get('dhcp') else 'Static',
            # A gatewayless NIC with an address is the machine link; the gateway
            # NIC is the corporate LAN.
            'role': 'Machine link' if (ipv4 and not gateway) else 'LAN',
            'link': _link_label(entry.get('linkspeedbps'),
                                entry.get('fullduplex')),
        })
    return success_response(rows)


# Columns for the two cards carved out of the old hand-written Network card on
# the Details tab. That card listed current addresses with a Primary badge and
# nothing else, which undersold what the rows actually hold.
_IPHISTORY_COLUMNS = [
    {'key': 'ipaddress', 'label': 'IP address'},
    {'key': 'macaddress', 'label': 'MAC'},
    {'key': 'status', 'label': 'Status'},
    {'key': 'firstseen', 'label': 'First seen'},
    {'key': 'lastchanged', 'label': 'Last changed'},
]


def _isostamp(value):
    """ISO-8601 with a Z suffix, matching every other timestamp this file emits.

    Formatting is left to the client so it can render in the viewer's locale
    and timezone; returning a preformatted string here would pin every reader
    to the server's.
    """
    return value.isoformat() + 'Z' if value else None


def _communication_rows(asset_id: int):
    """Every Communication row for an asset, newest change first.

    These rows are an ACCIDENTAL history rather than a designed one.
    _record_interfaces in the computers plugin matches an incoming NIC on MAC
    then on IP and inserts when neither matches, and it never deletes - so an
    address the machine has not held for a year is still sitting here next to
    the current one. That is genuinely useful (it answers "what was this box on
    before?"), so it is surfaced as history rather than cleaned away.
    """
    from shopdb.api import Communication

    # MySQL has no NULLS LAST, so sort on an explicit "is null" key first.
    # Rows that never got a modifieddate sink to the bottom on every dialect.
    return (Communication.query
            .filter_by(assetid=asset_id)
            .order_by(Communication.modifieddate.is_(None),
                      Communication.modifieddate.desc(),
                      Communication.communicationid.desc())
            .all())


def _iphistory_status(comm):
    """What this row IS, in the vocabulary somebody troubleshooting would use.

    Primary is the address the collector last picked as this box's main one.
    APIPA is called out by name because a 169.254.x.x address is not an address
    the machine was given - it is what Windows assigns when DHCP failed, so its
    presence in the list is a record of a past network fault, not a past
    subnet.
    """
    address = (comm.ipaddress or '').strip()
    if address.startswith('169.254.'):
        return 'APIPA (no DHCP)'
    if comm.isprimary:
        return 'Primary'
    if comm.ismachinenetwork:
        return 'Machine network'
    return 'Previous'


@endpoints_bp.route('/asset/<int:asset_id>/iphistory', methods=['GET'])
@require_permission('endpoints.view')
def panel_iphistory(asset_id: int):
    """Table panel: every address this asset has held, newest change first.

    Replaces the hand-written Network card that used to sit on the Details tab
    and show only the current addresses. Dates come from the row's own
    lifecycle columns, so they are labelled for what they are: 'Last changed'
    is when a field on the row last moved, NOT when the address was last seen.
    An address the machine still holds and reports unchanged does not update
    modifieddate, so calling that column 'Last seen' would be a lie the data
    cannot support.
    """
    rows = []
    for comm in _communication_rows(asset_id):
        address = (comm.ipaddress or '').strip()
        if not address:
            continue
        rows.append({
            'ipaddress': address,
            'macaddress': (comm.macaddress or '').strip() or '-',
            'status': _iphistory_status(comm),
            'firstseen': _isostamp(comm.createddate),
            'lastchanged': _isostamp(comm.modifieddate),
        })
    return success_response(rows)


@endpoints_bp.route('/asset/<int:asset_id>/applications', methods=['GET'])
@require_permission('endpoints.view')
def panel_applications(asset_id: int):
    """Table panel: the installed-software (SBOM) snapshot for this endpoint.

    Reads the endpointapplications rows the collector last wrote for this asset
    (a per-report snapshot, not history) and returns them sorted by name. This
    is the raw discovered inventory: appid is carried through but not yet linked
    to the applications catalog ("Managed Applications"), so no managed status
    is surfaced here. An endpoint with no reported SBOM yields no rows and the
    panel hides itself.
    """
    rows = (EndpointApplication.query
            .filter_by(assetid=asset_id)
            .order_by(db.func.lower(EndpointApplication.displayname))
            .all())
    return success_response([
        {
            'displayname': row.displayname,
            'displayversion': row.displayversion or '-',
            'publisher': row.publisher or '-',
        }
        for row in rows
    ])


# =============================================================================
# Dashboard
# =============================================================================

@endpoints_bp.route('/dashboard/summary', methods=['GET'])
@require_permission('endpoints.fleet')
def dashboard_summary():
    """High-level fleet counts for the endpoints dashboard."""
    total = db.session.query(EndpointState).join(
        Asset, Asset.assetid == EndpointState.assetid
    ).filter(Asset.isactive.is_(True)).count()

    hours = _quiet_hours()
    cutoff = _naive_utcnow() - timedelta(hours=hours)

    reporting = db.session.query(EndpointState).join(
        Asset, Asset.assetid == EndpointState.assetid
    ).filter(
        Asset.isactive.is_(True),
        EndpointState.lastreportedat.isnot(None),
        EndpointState.lastreportedat >= cutoff,
    ).count()

    # Counted separately from quiet. An endpoint that has NEVER reported is a
    # different problem from one that reported and stopped: the first is an
    # enrolment or key failure, the second is a machine that went down. Folding
    # them together (quiet = total - reporting) hid the enrolment failures.
    never = db.session.query(EndpointState).join(
        Asset, Asset.assetid == EndpointState.assetid
    ).filter(
        Asset.isactive.is_(True),
        EndpointState.lastreportedat.is_(None),
    ).count()

    return success_response({
        'total': total,
        'reporting': reporting,
        'quiet': total - reporting - never,
        'never': never,
        'quiethours': hours,
    })


@endpoints_bp.route('/dashboard/quiet', methods=['GET'])
@require_permission('endpoints.fleet')
def dashboard_quiet():
    """Endpoints that reported before, then went quiet past the window.

    Never-reported endpoints are excluded: those are records without an agent,
    not agents that stopped. Silence is the only signal - there is no heartbeat
    separate from the report.
    """
    hours = _quiet_hours()
    cutoff = _naive_utcnow() - timedelta(hours=hours)
    now = _naive_utcnow()

    from plugins.computers.models import Computer

    rows = (db.session.query(EndpointState, Computer.computerid)
            .join(Asset, Asset.assetid == EndpointState.assetid)
            .join(Computer, Computer.assetid == EndpointState.assetid)
            .filter(Asset.isactive.is_(True),
                    EndpointState.lastreportedat.isnot(None),
                    EndpointState.lastreportedat < cutoff)
            .all())

    out = [{
        'computerid': computerid,
        'assetid': s.assetid,
        'hostname': s.hostname,
        'lastreported': s.lastreportedat.isoformat() + 'Z',
        'quietdays': (now - s.lastreportedat).days,
    } for s, computerid in rows]

    out.sort(key=lambda r: -r['quietdays'])
    return success_response(out[:50])


@endpoints_bp.route('/dashboard/stability', methods=['GET'])
@require_permission('endpoints.fleet')
def dashboard_stability():
    """Endpoints with recent crash/shutdown/corruption events.

    Bounded to a fixed 48h window and the critical event catalog, capped at 50
    rows: cost is a range scan on eventtime filtered to a short id list, not a
    scan of all history. Joins through Computer so each row links to the PC
    detail page.
    """
    from plugins.computers.models import Computer

    if not CRITICAL_EVENT_IDS:
        return success_response([])

    cutoff = _naive_utcnow() - timedelta(hours=48)

    rows = (db.session.query(
                Computer.computerid,
                EndpointState.hostname,
                db.func.count(EndpointEvent.endpointeventid),
                db.func.max(EndpointEvent.eventtime),
            )
            .select_from(EndpointEvent)
            .join(Asset, Asset.assetid == EndpointEvent.assetid)
            .join(Computer, Computer.assetid == EndpointEvent.assetid)
            .outerjoin(EndpointState,
                       EndpointState.assetid == EndpointEvent.assetid)
            .filter(Asset.isactive.is_(True),
                    EndpointEvent.eventid.in_(CRITICAL_EVENT_IDS),
                    EndpointEvent.eventtime.isnot(None),
                    EndpointEvent.eventtime >= cutoff)
            .group_by(Computer.computerid, EndpointState.hostname)
            .all())

    out = [{
        'computerid': computerid,
        'hostname': hostname,
        'eventcount': count,
        'lastevent': lasttime.isoformat() + 'Z' if lasttime else None,
    } for computerid, hostname, count, lasttime in rows]

    out.sort(key=lambda r: -r['eventcount'])
    return success_response(out[:50])


def _quiet_hours() -> int:
    """Quiet-report window in hours, from a site setting, default 24."""
    hours = 24
    setting = Setting.query.filter_by(key='endpoints_quietreporthours').first()
    if setting and (setting.value or '').strip():
        try:
            hours = int(setting.value)
        except (TypeError, ValueError):
            pass
    return hours


# =============================================================================
# Endpoint agent key (UI-managed collector.ingest.endpoints kill switch)
# =============================================================================

# The single permission the endpoint agent key authorizes. A managed token
# scoped to ONLY this authorizes the endpoints collector and nothing else, so
# rotating/revoking it stops endpoint-agent traffic without touching the
# computers/geenforce collectors.
AGENT_KEY_SCOPE = 'collector.ingest.endpoints'
AGENT_KEY_NAME = 'Endpoint agent key'


def _agentkey_status_payload():
    """Current state of the endpoint agent key for the settings page.

    Reports whether an active managed key exists (and its non-secret metadata)
    plus whether the bootstrap .env fallback is still accepted. Never exposes a
    secret; the full secret is only ever returned once, at regenerate time.
    """
    tokens = list_service_tokens(AGENT_KEY_SCOPE)
    active = tokens[0] if tokens else None
    envfallback = Setting.get('plugin.endpoints.envkeyfallback', default=True)
    provisioned = Setting.get('plugin.endpoints.agentkeyprovisioned',
                              default=False)
    return {
        'provisioned': bool(provisioned) or active is not None,
        'envkeyfallback': bool(envfallback),
        'key': active.to_dict() if active is not None else None,
    }


@endpoints_bp.route('/agentkey', methods=['GET'])
@require_role('admin')
def get_agentkey():
    """Report the endpoint agent key state (no secret)."""
    return success_response(_agentkey_status_payload())


@endpoints_bp.route('/agentkey/regenerate', methods=['POST'])
@require_role('admin')
def regenerate_agentkey():
    """Mint a fresh endpoint agent key, revoking any previous one.

    Hard cutover: the old key is deactivated in the same call, and the .env
    fallback is switched off, so the new managed token becomes the sole
    credential and revoking it is a true kill switch. The full secret is
    returned ONCE here and never again.
    """
    try:
        token, secret = mint_service_token(
            AGENT_KEY_SCOPE, AGENT_KEY_NAME, current_user.userid)
    except PermissionError as error:
        return error_response(ErrorCodes.FORBIDDEN, str(error), http_code=403)

    Setting.set('plugin.endpoints.agentkeyprovisioned', True,
                valuetype='boolean', category='plugin.endpoints')
    Setting.set('plugin.endpoints.envkeyfallback', False,
                valuetype='boolean', category='plugin.endpoints')

    result = _agentkey_status_payload()
    # The secret appears here and NOWHERE else, ever. Not stored, not logged.
    result['secret'] = secret
    result['warning'] = ('Save this key now. It will not be shown again. '
                         'Configure it as the agent X-API-Key.')
    return success_response(result, message='Endpoint agent key generated',
                            http_code=201)


@endpoints_bp.route('/agentkey', methods=['DELETE'])
@require_role('admin')
def revoke_agentkey():
    """Revoke the endpoint agent key: stop all managed-token endpoint traffic.

    Leaves envkeyfallback off (the kill switch stays armed). To restore
    reporting an operator regenerates a new key.
    """
    revoked = revoke_service_tokens(AGENT_KEY_SCOPE)
    Setting.set('plugin.endpoints.agentkeyprovisioned', False,
                valuetype='boolean', category='plugin.endpoints')
    result = _agentkey_status_payload()
    result['revoked'] = revoked
    return success_response(result, message='Endpoint agent key revoked')


# =============================================================================
# Endpoint commands (WebUI -> agent, allowlisted actions only)
# =============================================================================
#
# Operators enqueue a named, server-allowlisted action for one endpoint (JWT +
# endpoints.command). The agent long-polls /commands/next with its collector
# token, claims the single next pending command, runs it locally, and reports
# the outcome to /commands/<id>/result. The wire never carries a command string
# - only an allowlisted action name - so a compromised WebUI cannot ask an
# endpoint to run something arbitrary.

# How long the agent claim poll is held open server-side before returning empty.
# Kept below the agent HttpClient timeout and typical proxy idle timeouts so a
# quiet fleet re-polls cleanly instead of tripping a read timeout.
#
# 25s is the value this endpoint is designed around; it was temporarily dropped
# to 5s under the blocking gthread worker pool, where a held connection pinned
# one of a worker's 8 OS threads. Measured then: 14 agents at hold=25 left a
# single worker with 8 of 9 threads parked in timed sleep and UI requests hashed
# to that worker waited 12-24s. Under gevent a held connection is a greenlet and
# the per-worker ceiling is --worker-connections (2000), not 8 threads, so an
# idle hold no longer costs anything a UI request needs.
#
# Restored to 25 because hold length sets the fleet's REQUEST RATE, not its
# concurrency: the agent re-polls immediately on a 204, so hold=5 is 5x the
# requests of hold=25, and TLS handshake CPU was the measured burst bottleneck
# (devnotes/gevent-sizing-simulation.md). 25 also restores the margin against
# the agent's MinIdlePollDuration=5s floor, which CommandWorker.IsIdlePollClose
# uses to tell a normal hold-window close apart from a transient failure; at
# hold=5 that margin was ~20ms.
#
# Constraint: gunicorn --timeout must stay above this or a healthy long-poll is
# killed mid-hold. docker-compose.yml sets 60.
_CLAIM_HOLD_SECONDS = 25
_CLAIM_POLL_INTERVAL = 1.0


def _pin_holds(command, descriptor):
    """Does the descriptor still carry the digest this command was queued for?

    True when nothing was pinned (actions that deliver no content), or when the
    descriptor is missing entirely - a vanished script or version already sends
    the command through with no descriptor so the agent fails closed, and that
    path is not a content swap.
    """
    if not command.pinnedsha256:
        return True
    if descriptor is None:
        return True
    return hmac.compare_digest(
        str(descriptor.get('sha256') or ''), command.pinnedsha256)


def _fail_on_pin_drift(command, action):
    """Refuse a claim whose content changed after the operator approved it."""
    command.status = EndpointCommand.STATUS_FAILED
    command.resultmessage = (
        'Refused: the {} content changed after this command was queued'
        .format(action))
    command.completedat = _naive_utcnow()
    db.session.commit()
    current_app.logger.warning(
        'Endpoint command %s (%s, asset %s) refused: pinned sha256 %s no '
        'longer matches the resolved descriptor',
        command.endpointcommandid, action, command.assetid,
        command.pinnedsha256)
    AuditLog.log('refused', 'EndpointCommand',
                 entityid=command.endpointcommandid,
                 entityname=action,
                 details='content digest changed after enqueue')
    db.session.commit()
    return ('', 204)


def _reap_stale_commands(asset_id):
    """Expire in-flight commands an agent will never finish for this endpoint.

    The queue is at-most-one-in-flight: a single PENDING or RUNNING row blocks
    every new command. If the agent is killed, uninstalled, or goes offline
    after a command is claimed (RUNNING) or before it claims one (PENDING), that
    row would pin the queue forever and every operator action returns 409. Reap
    rows past their staleness ceiling to STATUS_EXPIRED so the queue self-heals.
    Caller commits (this runs inside the enqueue transaction).
    """
    now = _naive_utcnow()
    running_cutoff = now - timedelta(seconds=EndpointCommand.RUNNING_STALE_SECONDS)
    pending_cutoff = now - timedelta(seconds=EndpointCommand.PENDING_STALE_SECONDS)

    stale = (EndpointCommand.query
             .filter(EndpointCommand.assetid == asset_id,
                     db.or_(
                         db.and_(
                             EndpointCommand.status == EndpointCommand.STATUS_RUNNING,
                             EndpointCommand.claimedat.isnot(None),
                             EndpointCommand.claimedat < running_cutoff),
                         db.and_(
                             EndpointCommand.status == EndpointCommand.STATUS_PENDING,
                             EndpointCommand.createdat < pending_cutoff)))
             .all())
    for command in stale:
        command.status = EndpointCommand.STATUS_EXPIRED
        command.completedat = now
        if not command.resultmessage:
            command.resultmessage = (
                'Expired: agent did not report a result before the staleness '
                'ceiling (agent offline, killed, or reinstalled mid-command).')
    return stale


def _collector_auth_or_error():
    """Authorize an agent-facing command route on the endpoints collector token.

    Reuses the collector auth path so the command channel shares the endpoint
    agent key (kill switch) and introduces no second secret.
    """
    return collector_authorized('endpoints')


def _resolve_asset_for_hostname(hostname):
    """Resolve a reported hostname to its asset id via the computers plugin."""
    from plugins.computers.models import Computer

    computer = (Computer.query
                .filter(db.func.lower(Computer.hostname) == hostname.lower())
                .first())
    return computer.assetid if computer is not None else None


_HOSTED_PACKAGE_PREFIX = '/api/applications/package/'


def _hosted_package_filename(application):
    """Return the stored filename for a ShopDB-hosted installer, or None.

    installpath is the core package URL (/api/applications/package/<filename>);
    a site-typed share path (not our upload) has no servable file here.
    """
    installpath = application.installpath or ''
    if not installpath.startswith(_HOSTED_PACKAGE_PREFIX):
        return None
    return installpath[len(_HOSTED_PACKAGE_PREFIX):]


def _backfill_app_installer_sha256(application):
    """Compute and persist installersha256 for a pre-migration package.

    Packages uploaded before the 7d36 migration have a null installersha256,
    which used to force a re-upload (Bug B). When the hosted package file is
    present on disk, hash it once, stamp the app, and commit so future resolves
    hit the stored value. Returns the digest, or None when no servable file.
    """
    filename = _hosted_package_filename(application)
    if not filename:
        return None
    package_dir = os.path.join(current_app.instance_path, 'applicationpackages')
    package_path = os.path.join(package_dir, filename)
    if not os.path.isfile(package_path):
        return None
    digest = hashlib.sha256()
    with open(package_path, 'rb') as package_file:
        for chunk in iter(lambda: package_file.read(65536), b''):
            digest.update(chunk)
    application.installersha256 = digest.hexdigest()
    db.session.commit()
    return application.installersha256


def _resolve_update_app(target):
    """Resolve an update_app target to a delivery descriptor.

    Accepts either an app-level or a version-level source, so a push works
    whether or not the site tracks AppVersion rows (the applications UI stores a
    single app-level installer; versioning is optional):

      - "app:<appid>"       - the app's own uploaded installer (installersha256)
      - "version:<id>"      - a specific AppVersion's pinned package
      - bare "<int>"        - legacy: treated as an appversionid

    Returns (descriptor, error_message). The descriptor carries what the agent
    needs to fetch and run the pinned installer: a collector-authenticated
    download URL, its SHA-256, and the silent-install switches. The agent
    fetches the bytes with its existing collector credential; no bytes here.
    """
    kind, _sep, raw = str(target or '').partition(':')
    if not raw:
        # No prefix: legacy bare appversionid.
        kind, raw = 'version', kind

    try:
        ident = int(raw)
    except (TypeError, ValueError):
        return None, 'update_app target must be app:<id>, version:<id>, or an appversionid'

    if kind == 'app':
        application = db.session.get(Application, ident)
        if application is None:
            return None, f'Unknown appid: {ident}'
        if not application.installpath:
            return None, (f'Application {ident} has no uploaded installer package')
        # Backfill installersha256 for packages uploaded before the 7d36 migration
        # so app-level delivery resolves without a re-upload (Bug B).
        sha256 = application.installersha256 or _backfill_app_installer_sha256(application)
        if not sha256:
            return None, (f'Application {ident} has no uploaded installer package')
        # Carry the real installer filename (with extension) on the delivery URL so
        # the agent's Path.GetExtension routes MSI through msiexec (Bug A rc2). The
        # trailing filename is cosmetic for the agent; the app_id drives the lookup.
        filename = _hosted_package_filename(application)
        installpath = '/api/endpoints/apppackage/app/{}'.format(application.appid)
        if filename:
            installpath = '{}/{}'.format(installpath, filename)
        return {
            'source': 'app',
            'appid': application.appid,
            'appversionid': None,
            'appname': application.appname,
            'version': None,
            'installpath': installpath,
            'sha256': sha256,
            'silentargs': None,
        }, None

    if kind == 'version':
        version = db.session.get(AppVersion, ident)
        if version is None:
            return None, f'Unknown appversionid: {ident}'
        application = db.session.get(Application, version.appid)
        if application is None or not application.installpath:
            return None, ('Application for appversionid {} has no installer package'
                          .format(ident))
        # PIN THE FILE THAT IS ACTUALLY SERVED.
        #
        # /api/endpoints/apppackage/<appversionid> resolves the version to its
        # APPLICATION and serves application.installpath - there is no
        # version-specific artifact behind that route. So version.sha256 pinned
        # a digest that need not describe the bytes delivered, and where it was
        # null the descriptor carried no digest at all and the agent had nothing
        # to verify against. Deriving it from the application, exactly as the
        # app branch does, makes the pin describe the delivery.
        sha256 = (application.installersha256
                  or _backfill_app_installer_sha256(application))
        if not sha256:
            return None, ('Application for appversionid {} has an installer '
                          'package with no recorded SHA-256, so the delivery '
                          'cannot be verified'.format(ident))

        # Carry the real filename so the agent's Path.GetExtension routes an MSI
        # through msiexec. Without it every MSI push failed, because this branch
        # ended the URL at the id and the extension was simply absent.
        filename = _hosted_package_filename(application)
        installpath = '/api/endpoints/apppackage/{}'.format(version.appversionid)
        if filename:
            installpath = '/api/endpoints/apppackage/app/{}/{}'.format(
                application.appid, filename)

        return {
            'source': 'version',
            'appversionid': version.appversionid,
            'appid': version.appid,
            'appname': application.appname,
            'version': version.version,
            'installpath': installpath,
            'sha256': sha256,
            'silentargs': version.silentargs,
        }, None

    return None, f'Unknown update_app target kind: {kind!r}'


def _resolve_run_script(target):
    """Resolve a run_script target to a script delivery descriptor.

    Accepts either "script:<scriptid>" or a bare script name. The descriptor
    carries a collector-authenticated download URL, the body's SHA-256, the
    per-script run context (system or user), and the optional folder to run in.
    The agent fetches the body with its credential and runs it; the body never
    crosses the wire.

    workingdirectory is sent even when empty so the field is always present in
    the claim: an agent that receives it and ignores it is then distinguishable
    from one that never saw it, which matters while agents of both vintages are
    in the field. errormode rides the same rule, and for the same reason.
    """
    kind, _sep, raw = str(target or '').partition(':')
    script = None
    if kind == 'script' and raw:
        try:
            script = db.session.get(Script, int(raw))
        except (TypeError, ValueError):
            return None, 'run_script target script:<id> must be numeric'
    else:
        # Bare name (the whole target, prefix included when there was no colon).
        name = (target or '').strip()
        if name:
            script = Script.query.filter_by(scriptname=name).first()

    if script is None:
        return None, f'Unknown script: {target!r}'
    if not script.filename or not script.sha256:
        return None, f'Script {script.scriptname!r} has no uploaded body'
    if not script.isactive:
        return None, f'Script {script.scriptname!r} is inactive'

    return {
        'scriptid': script.scriptid,
        'scriptname': script.scriptname,
        'installpath': '/api/endpoints/scriptbody/{}'.format(script.scriptid),
        'sha256': script.sha256,
        'runcontext': script.runcontext,
        'workingdirectory': script.workingdirectory or '',
        'errormode': script.errormode or 'default',
    }, None


@endpoints_bp.route('/<int:asset_id>/commands', methods=['GET'])
@require_permission('endpoints.view')
def list_endpoint_commands(asset_id: int):
    """Recent commands for one endpoint, newest first (operator view)."""
    rows = (EndpointCommand.query
            .filter_by(assetid=asset_id)
            .order_by(EndpointCommand.createdat.desc())
            .limit(20)
            .all())
    return success_response([row.to_dict() for row in rows])


@endpoints_bp.route('/<int:asset_id>/policyholds', methods=['GET'])
@require_permission('endpoints.view')
def list_endpoint_policy_holds(asset_id: int):
    """Active break-glass holds for one endpoint.

    Gated by endpoints.view, not by the admin role that places a hold. A hold
    suppresses reconciliation, so an operator watching a machine drift from its
    policy needs to see that the drift is deliberate, who authorized it and when
    it lapses. Hiding that from non-admins turns an audited override back into
    unexplained behavior.
    """
    return success_response(ACTIVE_POLICY_HOLDS(asset_id))


@endpoints_bp.route('/<int:asset_id>/holdableactions', methods=['GET'])
@require_permission('endpoints.view')
def list_endpoint_holdable_actions(asset_id: int):
    """Desired-mode actions on this endpoint, with current hold state.

    Feeds the admin hold picker. Read-gated rather than admin-gated to match the
    holds list: knowing WHICH actions this endpoint reconciles is the same class
    of provenance as knowing which are currently held.
    """
    return success_response(HOLDABLE_ACTIONS_FOR(asset_id))


@endpoints_bp.route('/<int:asset_id>/policyholds', methods=['POST'])
@require_permission('endpoints.command')
def place_endpoint_policy_hold(asset_id: int):
    """Deliberately suspend desired-state reconciliation for one or more actions.

    The same authority as break-glass, reached directly instead of as a side
    effect of a denied command: an admin who knows they are about to work on a
    machine should not have to trip a policy denial first to stop the reconciler
    fighting them.

    Accepts `actions` (a list) or `action` (a single name, kept so existing
    callers keep working). Every named action is validated BEFORE anything is
    written, and one bad name rejects the whole request. Partial success on a
    break-glass control is the wrong failure mode - "three of your five holds
    landed" is worthless to someone about to take a machine apart, and the
    missing two are discovered only when the reconciler undoes their work.

    One shared window covers every action in the request: holding five actions
    means doing one job on one machine, and that job has one duration.

    Admin-only via caller_is_role for the same reason as the override branch - a
    scoped personal token is a narrower caller than its owner, so a leaked agent
    token must not be able to switch off policy enforcement.
    """
    if not caller_is_role('admin'):
        return error_response(ErrorCodes.FORBIDDEN,
                              'Placing a policy hold requires the admin role',
                              http_code=403)

    payload = request.get_json(silent=True) or {}
    reason = (payload.get('reason') or '').strip()

    raw = payload.get('actions')
    if raw is None:
        raw = [payload.get('action')] if payload.get('action') else []
    if not isinstance(raw, list):
        return error_response(ErrorCodes.VALIDATION_ERROR,
                              'actions must be a list of action names')

    actions = []
    for entry in raw:
        name = (entry or '').strip() if isinstance(entry, str) else ''
        if name and name not in actions:
            actions.append(name)

    if not actions:
        return error_response(ErrorCodes.VALIDATION_ERROR,
                              'Select at least one action to hold')
    if not reason:
        return error_response(ErrorCodes.VALIDATION_ERROR,
                              'A hold requires a reason explaining why')

    for name in actions:
        if not is_allowed_action(name):
            return error_response(ErrorCodes.VALIDATION_ERROR,
                                  f'Unknown or disallowed action: {name!r}')

    # Only desired-mode actions are holdable. Accepting anything else would
    # write a row that suppresses nothing, so the UI would report a hold that
    # has no effect - worse than refusing. Checked for every action up front so
    # the request is rejected whole rather than half-applied.
    holdable = {entry['action'] for entry in HOLDABLE_ACTIONS_FOR(asset_id)}
    unholdable = [name for name in actions if name not in holdable]
    if unholdable:
        return error_response(
            ErrorCodes.VALIDATION_ERROR,
            f'Not under desired-state enforcement on this endpoint, so there '
            f'is nothing to hold: {", ".join(sorted(unholdable))}')

    minutes = payload.get('minutes')
    try:
        minutes = int(minutes) if minutes is not None else None
    except (TypeError, ValueError):
        return error_response(ErrorCodes.VALIDATION_ERROR,
                              'minutes must be a whole number')
    # Zero is legal and means indefinite: the hold stands until released by
    # hand. Negatives are not - they would place an already-expired hold, which
    # reads as success in the UI while suppressing nothing.
    if minutes is not None and minutes < 0:
        return error_response(ErrorCodes.VALIDATION_ERROR,
                              'minutes cannot be negative (use 0 for an '
                              'indefinite hold)')

    indefinite = minutes == 0
    window = 'indefinite' if indefinite else minutes

    audit_log(
        action='policyholdplaced',
        entitytype='Asset',
        entityid=asset_id,
        changes={
            'actions': actions,
            'reason': reason,
            'minutes': window,
            'indefinite': indefinite,
            'placedby': getattr(current_user, 'username', None),
        },
        details=(f'Policy hold placed on {", ".join(actions)} '
                 f'({window}): {reason}'),
    )

    for name in actions:
        placed = PLACE_POLICY_HOLD(
            assetid=asset_id,
            action=name,
            reason=reason,
            userid=getattr(current_user, 'userid', None),
            username=getattr(current_user, 'username', None),
            minutes=minutes,
        )
        if not placed:
            return error_response(ErrorCodes.VALIDATION_ERROR,
                                  'Policy holds are unavailable on this site')
    return success_response(ACTIVE_POLICY_HOLDS(asset_id))


@endpoints_bp.route('/<int:asset_id>/policyholds/<action>', methods=['DELETE'])
@require_permission('endpoints.command')
def release_endpoint_policy_hold(asset_id: int, action: str):
    """End a hold early, resuming reconciliation of one action.

    Admin-only on the same reasoning as placing one: releasing a hold turns
    policy enforcement back on for a machine someone may still be working on,
    which is the same authority in the opposite direction.
    """
    if not caller_is_role('admin'):
        return error_response(ErrorCodes.FORBIDDEN,
                              'Releasing a policy hold requires the admin role',
                              http_code=403)

    released = RELEASE_POLICY_HOLD(asset_id, action)
    if not released:
        return error_response(ErrorCodes.NOT_FOUND,
                              f'No active hold on {action!r} for this endpoint',
                              http_code=404)

    audit_log(
        action='policyholdreleased',
        entitytype='Asset',
        entityid=asset_id,
        changes={
            'action': action,
            'releasedby': getattr(current_user, 'username', None),
        },
        details=f'Policy hold released early on {action}',
    )
    return success_response(ACTIVE_POLICY_HOLDS(asset_id))


@endpoints_bp.route('/<int:asset_id>/lock', methods=['GET'])
@require_permission('endpoints.view')
def get_endpoint_lock(asset_id: int):
    """The endpoint lock standing on this endpoint, or null.

    Read-gated rather than admin-gated, and that is the point of the feature: an
    operator whose command was refused needs to see "locked by <admin>: mid
    repair" and self-serve. Hiding the lock from non-admins would leave them
    with a bare 403 and a ticket to file about a machine that is working fine.
    """
    lock = ACTIVE_LOCK_FOR(asset_id)
    return success_response(lock.to_dict() if lock else None)


@endpoints_bp.route('/<int:asset_id>/lock', methods=['POST'])
@require_permission('endpoints.command')
def place_endpoint_lock(asset_id: int):
    """Quarantine this endpoint against non-admin commands.

    Admin-only via caller_is_role rather than current_user.hasrole, matching the
    override and hold routes: a scoped personal token is a narrower caller than
    its owner, so a leaked agent token must not be able to quarantine machines.

    A reason is required. The refusal shown to a blocked operator quotes it
    verbatim, so an empty one produces a dead end - the operator learns only
    that they may not act, not who to ask or when it lifts.
    """
    if not caller_is_role('admin'):
        return error_response(ErrorCodes.FORBIDDEN,
                              'Locking an endpoint requires the admin role',
                              http_code=403)

    payload = request.get_json(silent=True) or {}
    reason = (payload.get('reason') or '').strip()
    if not reason:
        return error_response(ErrorCodes.VALIDATION_ERROR,
                              'A lock requires a reason explaining why')

    lock = PLACE_ENDPOINT_LOCK(
        assetid=asset_id,
        reason=reason,
        userid=getattr(current_user, 'userid', None),
        username=getattr(current_user, 'username', None),
    )

    audit_log(
        action='endpointlocked',
        entitytype='Asset',
        entityid=asset_id,
        changes={
            'reason': reason,
            'lockedby': getattr(current_user, 'username', None),
        },
        details=f'Endpoint locked: {reason}',
    )
    return success_response(lock.to_dict())


@endpoints_bp.route('/<int:asset_id>/lock', methods=['DELETE'])
@require_permission('endpoints.command')
def release_endpoint_lock(asset_id: int):
    """Lift the quarantine on this endpoint.

    Admin-only on the same reasoning as setting one: unlocking returns a machine
    someone may still be physically working inside to the whole operator pool,
    which is the same authority in the opposite direction.
    """
    if not caller_is_role('admin'):
        return error_response(ErrorCodes.FORBIDDEN,
                              'Unlocking an endpoint requires the admin role',
                              http_code=403)

    lock = RELEASE_ENDPOINT_LOCK(
        asset_id,
        userid=getattr(current_user, 'userid', None),
        username=getattr(current_user, 'username', None),
    )
    if lock is None:
        return error_response(ErrorCodes.NOT_FOUND,
                              'No active lock on this endpoint',
                              http_code=404)

    audit_log(
        action='endpointunlocked',
        entitytype='Asset',
        entityid=asset_id,
        changes={
            'lockedby': lock.lockedbyusername,
            'reason': lock.reason,
            'releasedby': getattr(current_user, 'username', None),
        },
        details=f'Endpoint unlocked (was locked: {lock.reason})',
    )
    return success_response(None)


@endpoints_bp.route('/<int:asset_id>/commands', methods=['POST'])
@require_permission('endpoints.command')
def enqueue_endpoint_command(asset_id: int):
    """Queue an allowlisted command for one endpoint.

    Rejects unknown actions, refuses non-admin callers while an endpoint lock
    stands, and refuses to stack a second pending/running command on the same
    endpoint, so the queue stays at-most-one-in-flight and the WebUI toggle
    cannot race itself.
    """
    payload = request.get_json(silent=True) or {}
    action = (payload.get('action') or '').strip()
    if not is_allowed_action(action):
        return error_response(ErrorCodes.VALIDATION_ERROR,
                              f'Unknown or disallowed action: {action!r}')

    # Endpoint lock: a quarantine on the CALLER axis, checked before the
    # permission and policy gates below so the refusal can name the lock, its
    # owner and its reason. Answering with a generic policy denial instead would
    # send an operator hunting through Configuration Policy for a cause that is
    # not there - the machine is fine, someone has simply reserved it.
    #
    # Admins pass straight through with no override argument and no reason. The
    # lock gates WHO may act, not WHICH actions exist, so there is no policy
    # decision here to break past; requiring break-glass would mean an admin had
    # to override their own team's quarantine to do the repair it was set for.
    lock = ACTIVE_LOCK_FOR(asset_id)
    if lock is not None and not caller_is_role('admin'):
        return error_response(
            ErrorCodes.FORBIDDEN,
            f'Endpoint locked by {lock.lockedbyusername or "a deleted user"}: '
            f'{lock.reason}',
            http_code=403)

    # Destructive actions (for example system_restart) demand a stricter grant
    # ON TOP OF endpoints.command, so a remote reboot cannot ride the routine
    # toggle permission.
    # caller_holds, not current_user.haspermission: the question is what THIS
    # caller may do, and a scoped personal token is a narrower caller than its
    # owner. Asking the owner let a token scoped to endpoints.command queue a
    # remote restart, a script run and an app deploy - the exact bypass
    # require_permission exists to prevent, reintroduced by checking by hand.
    elevated = elevated_permission_for(action)
    if elevated and not caller_holds(elevated):
        return error_response(
            ErrorCodes.FORBIDDEN,
            f'Action {action!r} requires the {elevated} permission',
            http_code=403)

    # Break-glass: an admin may push past a Configuration Policy denial, but
    # only by asking for it explicitly and saying why. The override is scoped to
    # POLICY gating alone - the elevated permission check above still stands, so
    # this cannot conjure a script run for someone who lacks endpoints.script.
    # It widens WHERE an action may run, never WHICH actions the caller may use.
    override = bool(payload.get('policyoverride'))
    overridereason = (payload.get('overridereason') or '').strip()

    if action not in ALLOWED_ACTIONS_FOR(assetid=asset_id):
        if not override:
            return error_response(
                ErrorCodes.VALIDATION_ERROR,
                f'Action not permitted by policy for this endpoint: {action!r}')
        # caller_is_role, not current_user.hasrole: a scoped personal token is a
        # narrower caller than its owner, so a leaked agent token must not
        # inherit break-glass from the admin who minted it.
        if not caller_is_role('admin'):
            return error_response(
                ErrorCodes.FORBIDDEN,
                'Overriding a policy denial requires the admin role',
                http_code=403)
        if not overridereason:
            return error_response(
                ErrorCodes.VALIDATION_ERROR,
                'An override requires overridereason explaining why')

        # Audited before the command is enqueued: if the enqueue fails after
        # this point, the record of the attempt still survives. A break-glass
        # that is only logged on success hides exactly the cases worth reviewing.
        # Structured context goes in changes (JSON-serialized by the model);
        # details is a plain Text column, so it carries the readable summary.
        audit_log(
            action='policyoverridden',
            entitytype='Asset',
            entityid=asset_id,
            changes={
                'action': action,
                'reason': overridereason,
                'overriddenby': getattr(current_user, 'username', None),
            },
            details=f'Policy denial overridden for {action}: {overridereason}',
        )

        # A gate-mode override is a one-shot; a desired-mode override is a fight
        # with the reconciler, which would re-enqueue the opposite command on the
        # next check-in and make the override look broken. Hold it open.
        PLACE_POLICY_HOLD(
            assetid=asset_id,
            action=action,
            reason=overridereason,
            userid=getattr(current_user, 'userid', None),
            username=getattr(current_user, 'username', None),
        )

    # Generic actions (service_* / process_*) must carry the subject to act on;
    # fixed-target actions (mtconnect_*) must not. Validate both directions so a
    # caller cannot send a targetless generic action (nothing to act on) or
    # smuggle a target onto a pinned action.
    target = (payload.get('target') or '').strip() or None
    targetkind = action_targetkind(action)
    if requires_target(action):
        if not target:
            return error_response(
                ErrorCodes.VALIDATION_ERROR,
                f'Action {action!r} requires a target')
    elif target is not None:
        return error_response(
            ErrorCodes.VALIDATION_ERROR,
            f'Action {action!r} does not accept a target')

    # update_app pins a specific application version; verify it resolves to a
    # real installer now so a broken push is rejected at enqueue rather than
    # failing opaquely on the endpoint minutes later.
    # Digest of the content this push was APPROVED against. Kept on the row so
    # the claim path can refuse a descriptor that has changed since; see
    # EndpointCommand.pinnedsha256.
    pinnedsha256 = None

    if action == 'update_app':
        descriptor, resolve_error = _resolve_update_app(target)
        if resolve_error is not None:
            return error_response(ErrorCodes.VALIDATION_ERROR, resolve_error)
        pinnedsha256 = (descriptor or {}).get('sha256')

    # run_script pins a ShopDB-hosted script; verify it resolves to a real
    # uploaded body now so a broken push is rejected at enqueue rather than
    # failing opaquely on the endpoint later.
    if action == 'run_script':
        script, script_error = _resolve_run_script(target)
        if script_error is not None:
            return error_response(ErrorCodes.VALIDATION_ERROR, script_error)
        pinnedsha256 = (script or {}).get('sha256')

    state = EndpointState.query.filter_by(assetid=asset_id).first()
    if state is None:
        return error_response(ErrorCodes.NOT_FOUND,
                              f'No endpoint state for asset {asset_id}',
                              http_code=404)

    # Self-heal a queue wedged by a command the agent will never finish (killed,
    # uninstalled, or offline mid-flight) before enforcing at-most-one-in-flight,
    # so a dead in-flight row cannot permanently 409 every operator action.
    _reap_stale_commands(asset_id)

    pending = (EndpointCommand.query
               .filter(EndpointCommand.assetid == asset_id,
                       EndpointCommand.status.in_((
                           EndpointCommand.STATUS_PENDING,
                           EndpointCommand.STATUS_RUNNING)))
               .first())
    if pending is not None:
        return error_response(
            ErrorCodes.VALIDATION_ERROR,
            'A command is already in flight for this endpoint',
            http_code=409)

    command = EndpointCommand(
        assetid=asset_id,
        action=action,
        target=target,
        targetkind=targetkind if target else None,
        status=EndpointCommand.STATUS_PENDING,
        requestedby=getattr(current_user, 'userid', None),
        pinnedsha256=pinnedsha256,
    )
    db.session.add(command)

    AuditLog.log('created', 'EndpointCommand',
                 entityname=action,
                 details=f'action={action} target={target} asset={asset_id}')

    db.session.commit()
    return success_response(command.to_dict(),
                            message='Command queued', http_code=201)


@endpoints_bp.route('/monitors/resolved', methods=['GET'])
def resolved_monitors_for_agent():
    """Agent watch list: the monitors this endpoint should report on.

    Collector-token auth (not JWT), same as the command channel. The agent
    calls this each telemetry cycle with its hostname and receives the process
    and service monitors that site policy (and the shipped defaults) attach to
    this endpoint's PC type, so a policy-attached monitor is enumerated and
    reported without editing appsettings.json on the box. The shape mirrors the
    agent's MonitorConfig (processes[]/services[] of {name, app}) so the agent
    can bind it directly: name is the wire identity (process image or service
    key), app is the operator label. An unknown hostname or a PC type with no
    monitors returns empty lists, which the agent merges as "watch nothing extra".
    """
    error = _collector_auth_or_error()
    if error is not None:
        return error

    hostname = (request.args.get('hostname') or '').strip()
    if not hostname:
        return error_response(ErrorCodes.VALIDATION_ERROR,
                              'hostname is required')

    asset_id = _resolve_asset_for_hostname(hostname)
    if asset_id is None:
        return error_response(ErrorCodes.NOT_FOUND,
                              f'Unknown hostname: {hostname!r}', http_code=404)

    processes = []
    services = []
    for monitor in _resolved_monitors_for_asset(asset_id):
        kind = (monitor.kind or '').lower()
        if kind == 'process':
            wirename = (monitor.processname or '').strip()
            bucket = processes
        elif kind == 'service':
            wirename = (monitor.servicename or '').strip()
            bucket = services
        else:
            continue
        if not wirename:
            continue
        entry = {
            'name': wirename,
            'app': (monitor.displayname or wirename),
        }
        # Only process monitors carry a launch path, and only when set. The
        # agent never trusts it blindly: it validates the path against an
        # approved-root allowlist before it becomes launchable (fail closed).
        if kind == 'process':
            launchpath = (monitor.launchpath or '').strip()
            if launchpath:
                entry['launchpath'] = launchpath
        bucket.append(entry)

    return success_response({'processes': processes, 'services': services})


@endpoints_bp.route('/commands/next', methods=['GET'])
def claim_next_command():
    """Agent long-poll: claim the next pending command for a hostname.

    Collector-token auth (not JWT). Holds the request open briefly and returns
    the single oldest pending command for the caller's endpoint, atomically
    marking it running so no other poll claims it. Returns 204 when nothing is
    queued within the hold window - the agent simply re-polls.
    """
    error = _collector_auth_or_error()
    if error is not None:
        return error

    hostname = (request.args.get('hostname') or '').strip()
    if not hostname:
        return error_response(ErrorCodes.VALIDATION_ERROR,
                              'hostname is required')

    asset_id = _resolve_asset_for_hostname(hostname)
    if asset_id is None:
        return error_response(ErrorCodes.NOT_FOUND,
                              f'Unknown hostname: {hostname!r}', http_code=404)

    deadline = time.monotonic() + _CLAIM_HOLD_SECONDS
    while True:
        command = (EndpointCommand.query
                   .filter_by(assetid=asset_id,
                              status=EndpointCommand.STATUS_PENDING)
                   .order_by(EndpointCommand.createdat.asc())
                   .first())
        if command is not None:
            command.status = EndpointCommand.STATUS_RUNNING
            command.claimedat = _naive_utcnow()
            db.session.commit()
            claim = {
                'endpointcommandid': command.endpointcommandid,
                'action': command.action,
                'target': command.target,
                'targetkind': command.targetkind,
            }
            # update_app rides a resolved delivery descriptor: the agent gets
            # the package URL, its pinned SHA-256, and the version's silent-
            # install switches, and fetches the bytes with its collector
            # credential. If the version was deleted between enqueue and claim,
            # send the command through with no descriptor and let the agent
            # fail closed rather than 500 the poll.
            if command.action == 'update_app':
                descriptor, _resolve_error = _resolve_update_app(command.target)
                if not _pin_holds(command, descriptor):
                    return _fail_on_pin_drift(command, 'update_app')
                if descriptor is not None:
                    claim['app'] = descriptor
            # run_script rides a resolved script descriptor: the agent gets a
            # collector-authenticated body URL, the pinned SHA-256, and the
            # per-script run context. Fetch happens agent-side with the
            # collector credential. If the script vanished between enqueue and
            # claim, send the command through with no descriptor and let the
            # agent fail closed rather than 500 the poll.
            if command.action == 'run_script':
                script_descriptor, _script_error = _resolve_run_script(command.target)
                if not _pin_holds(command, script_descriptor):
                    return _fail_on_pin_drift(command, 'run_script')
                if script_descriptor is not None:
                    claim['script'] = script_descriptor
            return success_response(claim)

        # Nothing queued: release the session so a concurrent enqueue is visible
        # on the next poll, wait a beat, then re-check until the hold expires.
        db.session.rollback()
        if time.monotonic() >= deadline:
            return ('', 204)
        time.sleep(_CLAIM_POLL_INTERVAL)


@endpoints_bp.route('/commands/<int:command_id>/result', methods=['POST'])
def report_command_result(command_id: int):
    """Agent reports a claimed command's outcome (collector-token auth).

    Finalizes the command row and, for MTConnect actions that observed the run
    state, caches it on the EndpointState so the WebUI toggle reflects reality
    between polls. Both writes land in one commit.
    """
    error = _collector_auth_or_error()
    if error is not None:
        return error

    command = db.session.get(EndpointCommand, command_id)
    if command is None:
        return error_response(ErrorCodes.NOT_FOUND,
                              f'No command {command_id}', http_code=404)

    # An agent may only finalize ITS OWN command. The collector key is shared
    # fleet-wide, so without this any agent could post a result for any command
    # id - marking an install succeeded on a machine that never ran it, which
    # writes a false compliance record. Binding to the claimed hostname is a
    # floor, not a fix: the credential still does not prove which endpoint is
    # calling, and only a per-endpoint credential closes that.
    # Optional because the shipped agent does not send it yet; enforced when it
    # does, so an updated fleet gets the check without a flag day.
    hostname = (request.args.get('hostname') or '').strip()
    if hostname:
        caller_asset_id = _resolve_asset_for_hostname(hostname)
        if caller_asset_id != command.assetid:
            return error_response(
                ErrorCodes.FORBIDDEN,
                'This command belongs to another endpoint', http_code=403)

    # Only a claimed command can be finalized. Needs no agent change, and it
    # closes two things the hostname check cannot: reporting on a command that
    # was never claimed, and overwriting a terminal result - which would let a
    # recorded failure be replaced by a success.
    if command.status != EndpointCommand.STATUS_RUNNING:
        return error_response(
            ErrorCodes.VALIDATION_ERROR,
            'Command {} is {}, not running'.format(
                command_id, command.status),
            http_code=409)

    payload = request.get_json(silent=True) or {}
    succeeded = bool(payload.get('succeeded'))
    command.status = (EndpointCommand.STATUS_SUCCEEDED if succeeded
                      else EndpointCommand.STATUS_FAILED)
    command.resultmessage = (payload.get('message') or '')[:2000] or None
    detail = payload.get('detail')
    command.resultjson = json.dumps(detail, separators=(',', ':')) \
        if detail is not None else None
    command.completedat = _naive_utcnow()

    # Cache the observed MTConnect state so the toggle shows a real value. The
    # agent sends mtconnectrunning on any mtconnect_* result that probed it.
    if 'mtconnectrunning' in payload:
        state = EndpointState.query.filter_by(assetid=command.assetid).first()
        if state is not None:
            running = payload.get('mtconnectrunning')
            state.mtconnectrunning = bool(running) if running is not None else None
            state.mtconnectcheckedat = _naive_utcnow()

    db.session.commit()
    return success_response(command.to_dict(), message='Result recorded')


@endpoints_bp.route('/apppackage/<int:appversionid>', methods=['GET'])
def download_app_package(appversionid: int):
    """Agent-facing installer download (collector-token auth).

    The core /api/applications/package route is gated on applications.view, a
    permission the endpoint agent's collector token does not hold. This route
    authorizes on the same endpoints collector token the agent already uses for
    telemetry and command polling, so the whole app-delivery path rides one key.
    Serves the exact bytes the update_app descriptor pinned by SHA-256.
    """
    from shopdb.api import send_upload

    error = _collector_auth_or_error()
    if error is not None:
        return error

    version = db.session.get(AppVersion, appversionid)
    if version is None:
        return error_response(ErrorCodes.NOT_FOUND,
                              f'Unknown appversionid: {appversionid}', http_code=404)
    application = db.session.get(Application, version.appid)
    if application is None or not application.installpath:
        return error_response(ErrorCodes.NOT_FOUND,
                              'No installer package for this application',
                              http_code=404)

    # installpath is the core package URL (/api/applications/package/<filename>);
    # take the stored filename and serve it from the core package directory. A
    # site-typed share path (not our upload) has no servable file here.
    prefix = '/api/applications/package/'
    if not application.installpath.startswith(prefix):
        return error_response(ErrorCodes.NOT_FOUND,
                              'Application installer is not a ShopDB-hosted package',
                              http_code=404)
    filename = application.installpath[len(prefix):]
    package_dir = os.path.join(current_app.instance_path, 'applicationpackages')
    return send_upload(package_dir, filename, as_attachment=True)


@endpoints_bp.route('/apppackage/app/<int:app_id>', methods=['GET'])
@endpoints_bp.route('/apppackage/app/<int:app_id>/<path:filename>', methods=['GET'])
def download_app_package_by_app(app_id: int, filename: str = None):
    """Agent-facing installer download by APP (collector-token auth).

    The app-level counterpart to download_app_package: serves the application's
    own uploaded installer (installpath + installersha256) so update_app can
    resolve an install source by app alone, with no AppVersion row. Same
    collector-token auth and same package directory as the per-version route.

    The optional trailing <filename> segment lets the delivery URL carry the
    real installer extension (so the agent routes MSI through msiexec); it is
    validated against the stored filename and does not drive the lookup.
    """
    from shopdb.api import send_upload

    error = _collector_auth_or_error()
    if error is not None:
        return error

    application = db.session.get(Application, app_id)
    if application is None or not application.installpath:
        return error_response(ErrorCodes.NOT_FOUND,
                              'No installer package for this application',
                              http_code=404)
    prefix = '/api/applications/package/'
    if not application.installpath.startswith(prefix):
        return error_response(ErrorCodes.NOT_FOUND,
                              'Application installer is not a ShopDB-hosted package',
                              http_code=404)
    stored_filename = application.installpath[len(prefix):]
    if filename is not None and filename != stored_filename:
        return error_response(ErrorCodes.NOT_FOUND,
                              'Requested installer filename does not match this application',
                              http_code=404)
    package_dir = os.path.join(current_app.instance_path, 'applicationpackages')
    return send_upload(package_dir, stored_filename, as_attachment=True)


@endpoints_bp.route('/scriptbody/<int:script_id>', methods=['GET'])
def download_script_body(script_id: int):
    """Agent-facing script body download (collector-token auth).

    The core /api/scripts/body route is gated on applications.view, which the
    endpoint agent's collector token does not hold. This route authorizes on
    the same endpoints collector token the agent already uses, keeping the
    whole script-delivery path on one key. Serves the exact bytes the
    run_script descriptor pinned by SHA-256.
    """
    from shopdb.api import send_upload

    error = _collector_auth_or_error()
    if error is not None:
        return error

    script = db.session.get(Script, script_id)
    if script is None or not script.filename:
        return error_response(ErrorCodes.NOT_FOUND,
                              'No body for this script', http_code=404)
    script_dir = os.path.join(current_app.instance_path, 'scripts')
    return send_upload(script_dir, script.filename, as_attachment=True)


# Operator issue reports (tray screenshot + free-text notes)
# ---------------------------------------------------------------------------
# Retention groundwork: how many days an operator report (and its screenshot
# file) is kept before the purge sweep removes it. Read from the Setting store
# so Agent Configuration can expose a toggle later without a code change; the
# default is 7 days. A value of 0 or less means "keep forever" (purge disabled).
OPERATOR_REPORT_RETENTION_SETTING = 'endpoints.operatorreport.retentiondays'
OPERATOR_REPORT_RETENTION_DEFAULT_DAYS = 7

# Screenshots are small PNGs from a virtual-screen grab. Cap the upload so a
# malformed or hostile agent cannot fill the disk with one request.
OPERATOR_REPORT_MAX_IMAGE_BYTES = 10 * 1024 * 1024
# The multipart envelope carries the other form fields and the part boundaries
# too, so content_length is legitimately larger than the image. Allow for that
# rather than rejecting a screenshot that is just under the cap.
_OPERATOR_REPORT_FORM_SLACK = 64 * 1024
# PNG signature. The route stores and serves the file as image/png, so this is
# the one thing it accepts.
_PNG_MAGIC = b'\x89PNG\r\n\x1a\n'


def _operator_report_dir():
    return os.path.join(current_app.instance_path, 'operatorreports')


def _operator_report_retention_days():
    """Configured retention window in whole days (default 7, <=0 disables)."""
    raw = Setting.get(OPERATOR_REPORT_RETENTION_SETTING)
    if raw is None:
        return OPERATOR_REPORT_RETENTION_DEFAULT_DAYS
    try:
        return int(raw)
    except (TypeError, ValueError):
        return OPERATOR_REPORT_RETENTION_DEFAULT_DAYS


def _purge_expired_operator_reports():
    """Delete operator reports older than the retention window, files and all.

    Best-effort and idempotent: called opportunistically on upload so the feed
    stays bounded without a scheduler. A retention of <=0 disables the sweep.
    """
    days = _operator_report_retention_days()
    if days <= 0:
        return

    cutoff = _naive_utcnow() - timedelta(days=days)
    stale = OperatorReport.query.filter(OperatorReport.createdat < cutoff).all()
    if not stale:
        return

    report_dir = _operator_report_dir()
    for report in stale:
        if report.imagefilename:
            path = os.path.join(report_dir, report.imagefilename)
            try:
                os.remove(path)
            except OSError:
                pass
        db.session.delete(report)
    db.session.commit()


@endpoints_bp.route('/operatorreports', methods=['POST'])
def upload_operator_report():
    """Agent uploads one operator issue report (collector-token auth).

    The tray captures the operator's free-text note and an optional screenshot,
    then hands them to the service, which posts here as multipart/form-data on
    the same endpoints collector token used for telemetry. The reported hostname
    resolves to an asset; an unresolved hostname is a validation error. The
    screenshot bytes are written to the operator-reports uploads dir and the row
    holds only metadata plus the stored filename. One retention sweep runs
    opportunistically so the feed stays bounded.
    """
    import uuid

    error = _collector_auth_or_error()
    if error is not None:
        return error

    # Refuse an oversized body BEFORE anything parses it. content_length is a
    # header, so it is readable without touching the body; request.form is not
    # - the first access to it makes werkzeug parse the whole multipart and
    # spool the file part to a temp file. An earlier version of this check sat
    # below that line and so ran after the thing it exists to prevent. The
    # value is caller-supplied, which is why this is an early out and not the
    # enforcement; the bounded read below is.
    declared = request.content_length or 0
    if declared > OPERATOR_REPORT_MAX_IMAGE_BYTES + _OPERATOR_REPORT_FORM_SLACK:
        return error_response(
            ErrorCodes.VALIDATION_ERROR,
            'screenshot exceeds the {} byte limit'.format(
                OPERATOR_REPORT_MAX_IMAGE_BYTES))

    hostname = (request.form.get('hostname') or '').strip()
    if not hostname:
        return error_response(ErrorCodes.VALIDATION_ERROR,
                              'hostname is required')

    asset_id = _resolve_asset_for_hostname(hostname)
    if asset_id is None:
        return error_response(
            ErrorCodes.VALIDATION_ERROR,
            'could not resolve hostname to a computer: {!r}'.format(hostname))

    reason = (request.form.get('reason') or 'Other (free text)').strip()[:128]
    details = (request.form.get('details') or '').strip() or None

    imagefilename = None
    imagemimetype = None
    imagebytesize = None

    upload = request.files.get('screenshot')
    if upload is not None and upload.filename:
        # Read one byte past the cap: enough to know it was exceeded, without
        # taking the whole of a body that lied about its length.
        data = upload.read(OPERATOR_REPORT_MAX_IMAGE_BYTES + 1)
        if len(data) > OPERATOR_REPORT_MAX_IMAGE_BYTES:
            return error_response(
                ErrorCodes.VALIDATION_ERROR,
                'screenshot exceeds the {} byte limit'.format(
                    OPERATOR_REPORT_MAX_IMAGE_BYTES))
        if data and not data.startswith(_PNG_MAGIC):
            # The stored file is served back as image/png. Storing whatever
            # arrived under that name meant the extension and the bytes could
            # disagree; the upload headers make the document case inert, but
            # nothing is gained by keeping a file we already know is not what it
            # claims.
            return error_response(ErrorCodes.VALIDATION_ERROR,
                                  'screenshot must be a PNG')
        if data:
            imagefilename = 'report-{}.png'.format(uuid.uuid4().hex)
            imagemimetype = 'image/png'
            imagebytesize = len(data)
            report_dir = _operator_report_dir()
            os.makedirs(report_dir, exist_ok=True)
            with open(os.path.join(report_dir, imagefilename), 'wb') as handle:
                handle.write(data)

    report = OperatorReport(
        assetid=asset_id,
        reason=reason,
        details=details,
        imagefilename=imagefilename,
        imagemimetype=imagemimetype,
        imagebytesize=imagebytesize,
        reportedhostname=hostname,
    )
    db.session.add(report)
    db.session.commit()

    # Keep the feed bounded without a scheduler. Best-effort: a purge failure
    # must not fail the upload the operator already completed.
    try:
        _purge_expired_operator_reports()
    except Exception:
        db.session.rollback()

    return success_response(report.to_dict(),
                            message='Operator report received', http_code=201)


@endpoints_bp.route('/panel/<int:asset_id>/operatorreports', methods=['GET'])
@require_permission('endpoints.operatorreports')
def panel_operator_reports(asset_id: int):
    """Operator Reports panel: this PC's recent operator issue reports.

    Newest first. Each row carries a hasimage flag so the viewer knows whether
    to request the screenshot from the serve route below. Runs a retention
    sweep opportunistically so a viewed PC does not accumulate stale rows.
    """
    try:
        _purge_expired_operator_reports()
    except Exception:
        db.session.rollback()

    reports = (OperatorReport.query
               .filter_by(assetid=asset_id)
               .order_by(OperatorReport.createdat.desc())
               .all())
    return success_response({'reports': [r.to_dict() for r in reports]})


@endpoints_bp.route('/operatorreports/<int:operatorreportid>/image',
                    methods=['GET'])
@require_permission('endpoints.operatorreports')
def serve_operator_report_image(operatorreportid: int):
    """Serve one operator report's screenshot.

    Requires a user JWT plus endpoints.operatorreports.
    This route, not the panel, is what actually protects the image: the panel
    only decides whether a tab is drawn, while the bytes are reachable by URL to
    anyone who can guess an id. Screenshots are an uncontrolled capture of the
    operator's screen, so this must never widen back to endpoints.view.
    """
    from shopdb.api import send_upload

    report = db.session.get(OperatorReport, operatorreportid)
    if report is None or not report.imagefilename:
        return error_response(ErrorCodes.NOT_FOUND,
                              'No screenshot for this report', http_code=404)
    return send_upload(_operator_report_dir(), report.imagefilename)