Report each Windows event once, and persist the inventory already sent

Curated events were re-read from the newest 25 entries per channel on every
collect, and the server appended whatever arrived. One real event was
therefore stored again on every cadence tick for as long as it stayed in
that window, so event counts measured how long a box had been up rather
than what had happened to it.

The agent now keeps a per-channel watermark of the last Windows event
record id it reported, persisted under ProgramData so a service restart
does not replay history, and sends eventrecordid with each event. The
server dedupes on (assetid, eventchannel, eventrecordid) behind a unique
index, so a replay from an older agent cannot double-store either.

Six inventory keys the agent had been sending all along - bios, disks,
domain, memory, osinfo, processor - were parsed, logged as unpersisted,
and dropped. They now have columns and are written on check-in, along
with user context, pending reboot, and the cpupercent/memorypercent/
idleseconds fields that were declared in the contract but always null.

Session-scoped data cannot be read from the agent's LocalSystem session 0
at all: mapped drives and user-owned listeners do not exist there. A probe
runs in the interactive user's session and hands results back through a
spool file. It is not yet producing output on a live endpoint; the column
and panel are in place and read empty until that is fixed.

Also bounds the user-context script wait. It was WaitForSingleObject with
INFINITE, so a hung script pinned the command worker permanently.

plugins/endpoints/migrations/versions/0012_endpoint_eventrecordid.py

@@ -0,0 +1,80 @@
+"""Dedup curated Windows events on the Windows event record id.
+
+Adds endpointevents.eventrecordid and a unique index over
+(assetid, eventchannel, eventrecordid).
+
+Why: the agent used to re-read the newest window of every curated channel on
+each cadence tick, and this ingest appended those rows without a dedup check, so
+one real event was stored once per report for as long as it stayed in that
+window - up to 96 identical rows a day. The agent now carries a per-channel
+watermark keyed on the Windows record id, which is monotonic and unique within a
+channel on a machine; this index is the server-side guarantee that survives an
+agent reinstall, a restored image, or two overlapping reports, none of which
+preserve the agent's watermark.
+
+Nullable on purpose: an agent older than this field reports no record id, and
+MySQL permits repeated NULLs inside a unique index, so those rows keep the
+previous append-only behavior instead of failing ingest.
+
+Existing rows are left alone. They all predate the column and so carry NULL,
+which cannot collide with the new index. Purging the historical duplicates is a
+separate, deliberate data cleanup - not something a schema migration should do
+irreversibly.
+
+Idempotent: both the column-add and the index-create check the live schema
+first, so a db.create_all() site and a re-run both stay clean on strict MySQL 8.
+"""
+from alembic import op
+
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'endpoints0012eventrecordid'
+down_revision = 'endpoints0011trayversion'
+branch_labels = None
+depends_on = None
+
+INDEX_NAME = 'idx_endpointevents_assetrecord'
+
+
+def _has_column(table_name, column_name):
+    from sqlalchemy import inspect
+
+    inspector = inspect(op.get_bind())
+    if not inspector.has_table(table_name):
+        return False
+    return any(col['name'] == column_name
+               for col in inspector.get_columns(table_name))
+
+
+def _has_index(table_name, index_name):
+    from sqlalchemy import inspect
+
+    inspector = inspect(op.get_bind())
+    if not inspector.has_table(table_name):
+        return False
+    return any(index['name'] == index_name
+               for index in inspector.get_indexes(table_name))
+
+
+def upgrade():
+    if not _has_column('endpointevents', 'eventrecordid'):
+        op.add_column(
+            'endpointevents',
+            sa.Column('eventrecordid', sa.BigInteger(), nullable=True))
+
+    if not _has_index('endpointevents', INDEX_NAME):
+        op.create_index(
+            INDEX_NAME,
+            'endpointevents',
+            ['assetid', 'eventchannel', 'eventrecordid'],
+            unique=True)
+
+
+def downgrade():
+    if _has_index('endpointevents', INDEX_NAME):
+        op.drop_index(INDEX_NAME, table_name='endpointevents')
+
+    if _has_column('endpointevents', 'eventrecordid'):
+        op.drop_column('endpointevents', 'eventrecordid')

plugins/endpoints/migrations/versions/0013_endpoint_inventory_columns.py

@@ -0,0 +1,93 @@
+"""Persist the hardware, OS, and session inventory the agent already reports.
+
+Adds to endpointstates: disksjson, biosjson, memoryjson, processorjson,
+osinfojson, usercontextjson, domain, and pendingreboot.
+
+Why now: the agent has been sending disks, bios, memory, processor, osinfo and
+domain on every report for several versions. The ingest logged them as unknown
+payload keys and threw them away, which was the intended staging step - the
+shapes are now proven against real endpoint traffic, so they get columns.
+usercontext and pendingreboot are new in this same change.
+
+Shape follows the existing servicesjson/portsjson convention: newest snapshot
+per endpoint as MEDIUMTEXT, not one row per sample. The agent resends the full
+structure every cycle and operators ask "what is in this PC now", not "what was
+in it in March". History would be pure write amplification on a fleet-sized
+table.
+
+domain and pendingreboot are real columns rather than JSON because they are
+single values the fleet view filters, groups, and alerts on; burying a scalar in
+a blob makes it unqueryable.
+
+All columns are nullable. Every existing row predates them, and there is no
+correct backfill value - a machine whose reboot state has never been reported is
+NOT the same as one confirmed to have no reboot pending, and inventing False
+would silently create a fleet of "healthy" endpoints.
+
+Idempotent: each column-add checks the live schema first, so a db.create_all()
+site and a re-run both stay clean on strict MySQL 8.
+"""
+from alembic import op
+
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'endpoints0013inventorycolumns'
+down_revision = 'endpoints0012eventrecordid'
+branch_labels = None
+depends_on = None
+
+TABLE_NAME = 'endpointstates'
+
+# MEDIUMTEXT. An installed-application or disk list on a busy engineering
+# workstation comfortably exceeds the 64KB TEXT ceiling, and a silent truncation
+# there would corrupt the JSON rather than merely shorten it.
+_SNAPSHOT_COLUMNS = (
+    'disksjson',
+    'biosjson',
+    'memoryjson',
+    'processorjson',
+    'osinfojson',
+    'usercontextjson',
+)
+
+
+def _has_column(table_name, column_name):
+    from sqlalchemy import inspect
+
+    inspector = inspect(op.get_bind())
+    if not inspector.has_table(table_name):
+        return False
+    return any(col['name'] == column_name
+               for col in inspector.get_columns(table_name))
+
+
+def upgrade():
+    for column_name in _SNAPSHOT_COLUMNS:
+        if not _has_column(TABLE_NAME, column_name):
+            op.add_column(
+                TABLE_NAME,
+                sa.Column(column_name, sa.Text(length=16777215), nullable=True))
+
+    if not _has_column(TABLE_NAME, 'domain'):
+        op.add_column(
+            TABLE_NAME,
+            sa.Column('domain', sa.String(length=255), nullable=True))
+
+    if not _has_column(TABLE_NAME, 'pendingreboot'):
+        op.add_column(
+            TABLE_NAME,
+            sa.Column('pendingreboot', sa.Boolean(), nullable=True))
+
+
+def downgrade():
+    if _has_column(TABLE_NAME, 'pendingreboot'):
+        op.drop_column(TABLE_NAME, 'pendingreboot')
+
+    if _has_column(TABLE_NAME, 'domain'):
+        op.drop_column(TABLE_NAME, 'domain')
+
+    for column_name in reversed(_SNAPSHOT_COLUMNS):
+        if _has_column(TABLE_NAME, column_name):
+            op.drop_column(TABLE_NAME, column_name)