Make main the lean product a site receives

main was 16 days and 259 commits behind the line the product is actually built
on, at contract 0.30.0 against 0.35.0, with no per-PC credential code at all. It
was neither current nor curated - just stale. This brings it forward and, at the
same time, makes it mean something: main is what a receiving site installs, and
nothing else.

WHAT IT CARRIES. Twelve plugins - backups, computers, endpoints, guides,
knowledgebase, machines, network, notifications, printers, slides, tools,
warranty - plus applications, which is manifest-less core and always ships. The
only manifest dependency inside that set is endpoints on computers, and both are
in it, so the closure is self-contained.

WHAT IT DOES NOT. employees, geenforce, measuringtools, printedparts and usb
stay on the site line. So do the Windows installer tree, the VM rehearsal
helpers, the agent tooling config, one site's data loader, the internal audit
and findings records, and the planner spreadsheet. 1,135 tracked files against
1,438.

A REMOVAL THAT BREAKS THE APP IS NOT A REMOVAL, and two nearly shipped:

deploy/site-profile-universal.json still named all seventeen plugins. It is the
file docs/DEPLOY.md tells a site to provision from, so a fresh install would
have failed at `flask plugin apply-profile` with PluginError - the first thing a
receiving site does, broken by the branch meant to serve them. It now names the
twelve that exist, and its comment says the two must agree.

agent/ was never refreshed. Whoever populated this branch copied shopdb/,
plugins/, frontend/ and tests/ from the site line and left the agent tree at its
September state: 22 C# and PowerShell sources missing, including AgentCredentials,
EnrollmentService, AgentHttpFactory and the fleet-alert client, plus one file the
site line had deleted still present. The old tree is internally consistent, so it
compiles, and nothing in CI builds it - every gate was green over a server that
speaks enrollment and per-PC credentials to a bundled agent source that cannot.
Resynced whole.

DECISIONS THIS ENCODES, rather than defers:

- The plugin-authoring lab builds printedparts, whose manifest depends on
  employees, so it could not stay. The lab remains on the site line and main
  ships no build-along tutorial. That is a real loss for an adopting developer
  and is recorded here rather than discovered later.
- Recognition, Training and Recertification notifications are no longer seeded.
  Each splits one notification into a card per employee and renders a photo,
  which needs a plugin this branch does not bundle. Seeding them would offer an
  author a type that cannot resolve anybody. A site that installs employees gets
  them from that plugin's own seed.
- Windows/IIS is documented as the legacy path throughout, and the README leads
  with Docker, because that is what a receiving site gets.

The API description is regenerated for THIS branch - 347 paths and 483
operations, down from 415 and 573 - so the published contract describes what is
here rather than what the site line serves. docs/CONFIG.md lost the two
measuring-tool label keys the code no longer declares, which two lanes had
decided opposite ways.

Suite: 3124 passed, 6 skipped. Naming and style gates pass. The docs gates run
rather than skip, and every index line count is derived from the file it names.

scripts/loadtest/Dockerfile.gevent

@@ -0,0 +1,9 @@
+# Sizing-simulation overlay. NOT part of the shipped image.
+#
+# Layers gevent onto the built production image so the gevent worker model can
+# be measured against the gthread baseline under identical application code.
+# Delete with the rest of scripts/loadtest once sizing questions are settled.
+FROM shopdb-flask-wilm-api:latest
+
+USER root
+RUN pip install --no-cache-dir "gevent==25.9.1"

scripts/loadtest/burst_arrival.py

@@ -0,0 +1,219 @@
+"""Patch-day burst simulator: synchronized cold start of an endpoint fleet.
+
+WHY this is separate from mixed_workload.py: that harness RAMPS agents in over
+~20s and measures steady state, which deliberately avoids a thundering herd.
+Patch day is the opposite case and stresses different machinery:
+
+  - a fresh TLS handshake per agent, all at once (handshakes are CPU-heavy)
+  - _resolve_asset_for_hostname runs on every arrival, so the DB pool is hit by
+    a synchronized burst rather than the smooth churn of steady state
+  - no warm connections in the pool, no warm caches
+
+Modelled on the real agent (verified in agent/ShopDbEndpointAgent.Core/):
+  - CommandWorker.ExecuteAsync polls IMMEDIATELY, no startup delay, no jitter
+  - TelemetryWorker also reports immediately; its jitter applies only BETWEEN
+    runs, so first contact is synchronized too. Patch day is therefore TWO
+    simultaneous bursts per endpoint, not one.
+  - transient failures retry at a FIXED 10s (TransientRetryDelay), not
+    exponential-with-jitter, so a herd that fails together re-arrives together
+
+--arrivalseconds is the spread of the reboot window. 0 means every agent hits at
+once (single maintenance window, worst case); 60 approximates a staged ring.
+
+Stdlib asyncio only, so it needs no dependency install in the container.
+
+Usage:
+    python burst_arrival.py --fleet 700 --arrivalseconds 0 --seconds 120
+"""
+import argparse
+import asyncio
+import json
+import os
+import random
+import ssl
+import statistics
+import time
+
+STOP = False
+
+
+def percentile(values, fraction):
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    index = min(int(len(ordered) * fraction), len(ordered) - 1)
+    return ordered[index]
+
+
+async def request(host, port, context, method, path, timeout,
+                  api_key=None, body=None):
+    """One HTTP/1.1 request over a FRESH connection, so each call pays a full
+    TLS handshake - which is the point on patch day."""
+    started = time.monotonic()
+    writer = None
+    try:
+        reader, writer = await asyncio.wait_for(
+            asyncio.open_connection(host, port, ssl=context), timeout=timeout)
+        handshake = time.monotonic() - started
+        headers = [
+            '%s %s HTTP/1.1' % (method, path),
+            'Host: %s' % host,
+            'Accept: application/json',
+            'Connection: close',
+        ]
+        if api_key:
+            headers.append('X-API-Key: %s' % api_key)
+        payload = b''
+        if body is not None:
+            payload = json.dumps(body).encode()
+            headers.append('Content-Type: application/json')
+            headers.append('Content-Length: %d' % len(payload))
+        writer.write(('\r\n'.join(headers) + '\r\n\r\n').encode() + payload)
+        await writer.drain()
+        header = await asyncio.wait_for(reader.readuntil(b'\r\n'), timeout=timeout)
+        status = int(header.split()[1])
+        await asyncio.wait_for(reader.read(), timeout=timeout)
+        return status, time.monotonic() - started, handshake
+    except Exception as error:
+        return type(error).__name__, time.monotonic() - started, None
+    finally:
+        if writer is not None:
+            try:
+                writer.close()
+            except Exception:
+                pass
+
+
+def record(results, bucket, status, elapsed, handshake=None, first=False):
+    entry = results.setdefault(bucket, {
+        'times': [], 'status': {}, 'handshakes': [], 'firsttimes': []})
+    entry['times'].append(elapsed * 1000.0)
+    if handshake is not None:
+        entry['handshakes'].append(handshake * 1000.0)
+    if first:
+        entry['firsttimes'].append(elapsed * 1000.0)
+    key = str(status)
+    entry['status'][key] = entry['status'].get(key, 0) + 1
+
+
+async def agent(index, host, port, context, api_key, results, arrivalseconds):
+    """One endpoint coming back from a patch reboot.
+
+    Fires its telemetry report and its first command poll at the same instant,
+    mirroring the two workers both starting with no delay.
+    """
+    if arrivalseconds > 0:
+        await asyncio.sleep(random.uniform(0, arrivalseconds))
+
+    hostname = 'LOADTEST%04d' % (index + 1)
+
+    async def telemetry():
+        status, elapsed, handshake = await request(
+            host, port, context, 'POST', '/api/collector/heartbeat',
+            timeout=120.0, api_key=api_key, body={'hostname': hostname})
+        record(results, 'telemetry_firstcontact', status, elapsed, handshake,
+               first=True)
+
+    telemetry_task = asyncio.ensure_future(telemetry())
+
+    path = '/api/endpoints/commands/next?hostname=%s' % hostname
+    first = True
+    while not STOP:
+        status, elapsed, handshake = await request(
+            host, port, context, 'GET', path, timeout=120.0, api_key=api_key)
+        record(results, 'fleet', status, elapsed, handshake, first=first)
+        if first:
+            first = False
+        if not isinstance(status, int):
+            # Mirrors CommandWorker TransientRetryDelay: fixed 10s, no jitter,
+            # so a herd that fails together re-arrives together.
+            await asyncio.sleep(10.0)
+        else:
+            await asyncio.sleep(0.05)
+
+    await telemetry_task
+
+
+async def ui_user(host, port, context, results, think_seconds):
+    while not STOP:
+        status, elapsed, handshake = await request(
+            host, port, context, 'GET', '/api/dashboard/health', timeout=120.0)
+        record(results, 'ui', status, elapsed, handshake)
+        await asyncio.sleep(think_seconds)
+
+
+async def main():
+    global STOP
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--fleet', type=int, default=700)
+    parser.add_argument('--users', type=int, default=5)
+    parser.add_argument('--arrivalseconds', type=float, default=0.0)
+    parser.add_argument('--seconds', type=int, default=120)
+    parser.add_argument('--host', default='api')
+    parser.add_argument('--port', type=int, default=5001)
+    parser.add_argument('--think', type=float, default=3.0)
+    arguments = parser.parse_args()
+
+    api_key = os.environ.get('LOADTEST_TOKEN', '')
+    if not api_key:
+        raise SystemExit('LOADTEST_TOKEN not set')
+
+    context = ssl.create_default_context()
+    context.check_hostname = False
+    context.verify_mode = ssl.CERT_NONE
+
+    results = {}
+    tasks = []
+    for _ in range(arguments.users):
+        tasks.append(asyncio.ensure_future(ui_user(
+            arguments.host, arguments.port, context, results,
+            arguments.think)))
+    # Let the UI baseline establish before the storm hits.
+    await asyncio.sleep(5.0)
+
+    burst_started = time.monotonic()
+    for index in range(arguments.fleet):
+        tasks.append(asyncio.ensure_future(agent(
+            index, arguments.host, arguments.port, context, api_key, results,
+            arguments.arrivalseconds)))
+
+    await asyncio.sleep(arguments.seconds)
+    STOP = True
+    await asyncio.sleep(2.0)
+    for task in tasks:
+        task.cancel()
+
+    summary = {
+        'fleet': arguments.fleet,
+        'arrivalseconds': arguments.arrivalseconds,
+        'users': arguments.users,
+        'seconds': arguments.seconds,
+        'classes': {},
+    }
+    for bucket, entry in sorted(results.items()):
+        times = entry['times']
+        block = {
+            'samples': len(times),
+            'status': entry['status'],
+            'median_ms': round(statistics.median(times), 1) if times else 0.0,
+            'p95_ms': round(percentile(times, 0.95), 1),
+            'max_ms': round(max(times), 1) if times else 0.0,
+        }
+        if entry['handshakes']:
+            block['handshake_median_ms'] = round(
+                statistics.median(entry['handshakes']), 1)
+            block['handshake_p95_ms'] = round(
+                percentile(entry['handshakes'], 0.95), 1)
+            block['handshake_max_ms'] = round(max(entry['handshakes']), 1)
+        if entry['firsttimes']:
+            block['firstcontact_median_ms'] = round(
+                statistics.median(entry['firsttimes']), 1)
+            block['firstcontact_p95_ms'] = round(
+                percentile(entry['firsttimes'], 0.95), 1)
+            block['firstcontact_max_ms'] = round(max(entry['firsttimes']), 1)
+        summary['classes'][bucket] = block
+    print(json.dumps(summary, indent=2))
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

scripts/loadtest/force_pool.py

@@ -0,0 +1,22 @@
+"""Force SQLAlchemy pool sizing for a pool-comparison run.
+
+WHY this exists: shopdb/config.py does not read pool size from the environment,
+so a test container cannot be pointed at a different pool without a code change.
+This patches Config.SQLALCHEMY_ENGINE_OPTIONS at import time, before create_app
+builds the engine, purely so a bigger pool can be MEASURED before deciding
+whether to make it permanent.
+
+Not for production use. Mounted into a throwaway container as sitecustomize.py,
+which Python imports automatically at startup.
+"""
+import os
+
+from shopdb.config import Config
+
+_size = int(os.environ.get('FORCE_POOL_SIZE', '5'))
+_overflow = int(os.environ.get('FORCE_MAX_OVERFLOW', '10'))
+
+_options = dict(Config.SQLALCHEMY_ENGINE_OPTIONS)
+_options['pool_size'] = _size
+_options['max_overflow'] = _overflow
+Config.SQLALCHEMY_ENGINE_OPTIONS = _options

scripts/loadtest/loadtest_token.py

@@ -0,0 +1,62 @@
+"""Mint a temporary collector token for the load test, or revoke it after.
+
+WHY a minted token: plugin.endpoints.envkeyfallback is false, which is the
+agent kill switch working as intended - the env key no longer authorizes. The
+harness therefore needs a real managed token scoped collector.ingest.endpoints,
+exactly like a live agent.
+
+The token is named LOADTEST so revoke() removes only this one and never touches
+the real agent keys.
+
+Usage (inside the api container, with PYTHONPATH=/app and cwd /app):
+    python loadtest_token.py mint    # prints the secret once
+    python loadtest_token.py revoke
+"""
+import sys
+
+from shopdb import create_app, db
+from shopdb.core.models.apitoken import ApiToken
+
+TOKEN_NAME = 'LOADTEST temporary collector token'
+SCOPE = 'collector.ingest.endpoints'
+
+
+def mint():
+    existing = ApiToken.query.filter_by(name=TOKEN_NAME).first()
+    if existing is not None:
+        db.session.delete(existing)
+        db.session.commit()
+    owner = ApiToken.query.filter(
+        ApiToken.scopes.like('%' + SCOPE + '%')).first()
+    if owner is None:
+        raise SystemExit('no existing collector token to copy ownership from')
+    secret = ApiToken.generate_secret()
+    token = ApiToken(
+        userid=owner.userid,
+        name=TOKEN_NAME,
+        tokenprefix=ApiToken.prefix_of(secret),
+        tokenhash=ApiToken.hash_secret(secret),
+    )
+    token.scopelist = [SCOPE]
+    db.session.add(token)
+    db.session.commit()
+    print(secret)
+
+
+def revoke():
+    removed = ApiToken.query.filter_by(name=TOKEN_NAME).delete()
+    db.session.commit()
+    print('revoked %d loadtest token(s)' % removed)
+
+
+if __name__ == '__main__':
+    action = sys.argv[1] if len(sys.argv) > 1 else ''
+    app = create_app()
+    with app.app_context():
+        if action == 'mint':
+            mint()
+        elif action == 'revoke':
+            revoke()
+        else:
+            print(__doc__)
+            sys.exit(2)

scripts/loadtest/mixed_workload.py

@@ -0,0 +1,218 @@
+"""Mixed-workload generator for sizing simulations.
+
+WHY this shape: sizing cannot be judged from long-poll agents alone. The four
+traffic classes have different concurrency costs, and only their SUM decides
+whether a worker model holds:
+
+  fleet       long-poll agents, one continuously-held connection each
+  ui          human users clicking, latency-sensitive, the actual symptom
+  integration Maximo / ServiceNow style collector clients on a fixed interval
+  kiosk       unauthenticated display endpoints, public routes, low rate
+
+The fleet class is what saturates a thread pool; the ui class is what a person
+notices. Measuring them together is the point - a model that keeps agents happy
+while starving the UI has failed even though every agent request succeeded.
+
+Runs INSIDE a container on the compose network. Generating from the Windows host
+would put host ephemeral-port and TLS-handshake limits in the measurement path,
+which would look like server saturation but would not be.
+
+Stdlib asyncio only, so it needs no dependency install in the container.
+
+Usage:
+    python mixed_workload.py --fleet 300 --users 5 --seconds 120
+"""
+import argparse
+import asyncio
+import json
+import os
+import ssl
+import statistics
+import time
+
+STOP = False
+
+
+def percentile(values, fraction):
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    index = min(int(len(ordered) * fraction), len(ordered) - 1)
+    return ordered[index]
+
+
+async def request(host, port, context, method, path, timeout,
+                  api_key=None, bearer=None, body=None):
+    """One HTTP/1.1 request over a fresh connection. Returns (status, seconds)."""
+    started = time.monotonic()
+    writer = None
+    try:
+        reader, writer = await asyncio.wait_for(
+            asyncio.open_connection(host, port, ssl=context), timeout=timeout)
+        headers = [
+            '%s %s HTTP/1.1' % (method, path),
+            'Host: %s' % host,
+            'Accept: application/json',
+            'Connection: close',
+        ]
+        if api_key:
+            headers.append('X-API-Key: %s' % api_key)
+        if bearer:
+            headers.append('Authorization: Bearer %s' % bearer)
+        payload = b''
+        if body is not None:
+            payload = json.dumps(body).encode()
+            headers.append('Content-Type: application/json')
+            headers.append('Content-Length: %d' % len(payload))
+        blob = ('\r\n'.join(headers) + '\r\n\r\n').encode() + payload
+        writer.write(blob)
+        await writer.drain()
+        header = await asyncio.wait_for(reader.readuntil(b'\r\n'), timeout=timeout)
+        status = int(header.split()[1])
+        await asyncio.wait_for(reader.read(), timeout=timeout)
+        return status, time.monotonic() - started
+    except Exception as error:
+        return type(error).__name__, time.monotonic() - started
+    finally:
+        if writer is not None:
+            try:
+                writer.close()
+            except Exception:
+                pass
+
+
+def record(results, bucket, status, elapsed):
+    entry = results.setdefault(bucket, {'times': [], 'status': {}})
+    entry['times'].append(elapsed * 1000.0)
+    key = str(status)
+    entry['status'][key] = entry['status'].get(key, 0) + 1
+
+
+async def fleet_agent(index, host, port, context, api_key, results):
+    """Mirrors CommandWorker.cs: re-poll immediately after each response."""
+    hostname = 'LOADTEST%04d' % (index + 1)
+    path = '/api/endpoints/commands/next?hostname=%s' % hostname
+    while not STOP:
+        status, elapsed = await request(
+            host, port, context, 'GET', path, timeout=120.0, api_key=api_key)
+        record(results, 'fleet', status, elapsed)
+        await asyncio.sleep(0.05)
+
+
+async def ui_user(host, port, context, bearer, results, think_seconds):
+    """A human clicking around. Latency here is the user-visible symptom."""
+    paths = ['/api/dashboard/health', '/api/endpoints?page=1&per_page=25',
+             '/api/assets?page=1&per_page=25']
+    cursor = 0
+    while not STOP:
+        path = paths[cursor % len(paths)]
+        cursor += 1
+        # Unauthenticated probe still measures queueing; auth only changes the
+        # handler cost, not whether a worker slot was available to accept it.
+        status, elapsed = await request(
+            host, port, context, 'GET', path, timeout=120.0, bearer=bearer)
+        record(results, 'ui', status, elapsed)
+        await asyncio.sleep(think_seconds)
+
+
+async def integration_client(name, host, port, context, api_key, results,
+                             interval):
+    """Maximo / ServiceNow style poller on a fixed interval."""
+    while not STOP:
+        status, elapsed = await request(
+            host, port, context, 'GET', '/api/collector/status',
+            timeout=120.0, api_key=api_key)
+        record(results, 'integration', status, elapsed)
+        await asyncio.sleep(interval)
+
+
+async def kiosk_endpoint(index, host, port, context, results, interval):
+    """Unauthenticated display screen. Public route, low rate."""
+    path = '/api/dashboarddefaults/display-role'
+    while not STOP:
+        status, elapsed = await request(
+            host, port, context, 'GET', path, timeout=120.0)
+        record(results, 'kiosk', status, elapsed)
+        await asyncio.sleep(interval)
+
+
+async def main():
+    global STOP
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--fleet', type=int, default=300)
+    parser.add_argument('--users', type=int, default=5)
+    parser.add_argument('--integrations', type=int, default=2)
+    parser.add_argument('--kiosks', type=int, default=8)
+    parser.add_argument('--seconds', type=int, default=120)
+    parser.add_argument('--host', default='api')
+    parser.add_argument('--port', type=int, default=5001)
+    parser.add_argument('--think', type=float, default=3.0)
+    parser.add_argument('--integration-interval', type=float, default=30.0)
+    parser.add_argument('--kiosk-interval', type=float, default=15.0)
+    parser.add_argument('--rampseconds', type=float, default=20.0)
+    arguments = parser.parse_args()
+
+    api_key = os.environ.get('LOADTEST_TOKEN', '')
+    if not api_key:
+        raise SystemExit('LOADTEST_TOKEN not set')
+    bearer = os.environ.get('LOADTEST_JWT') or None
+
+    context = ssl.create_default_context()
+    context.check_hostname = False
+    context.verify_mode = ssl.CERT_NONE
+
+    results = {}
+    tasks = []
+
+    # Ramp the fleet in rather than opening every connection at once. A
+    # thundering herd measures handshake burst capacity, not steady state.
+    stagger = arguments.rampseconds / max(arguments.fleet, 1)
+
+    async def staggered(index):
+        await asyncio.sleep(index * stagger)
+        await fleet_agent(index, arguments.host, arguments.port, context,
+                          api_key, results)
+
+    for index in range(arguments.fleet):
+        tasks.append(asyncio.ensure_future(staggered(index)))
+    for _ in range(arguments.users):
+        tasks.append(asyncio.ensure_future(ui_user(
+            arguments.host, arguments.port, context, bearer, results,
+            arguments.think)))
+    for index in range(arguments.integrations):
+        tasks.append(asyncio.ensure_future(integration_client(
+            'integration%d' % index, arguments.host, arguments.port, context,
+            api_key, results, arguments.integration_interval)))
+    for index in range(arguments.kiosks):
+        tasks.append(asyncio.ensure_future(kiosk_endpoint(
+            index, arguments.host, arguments.port, context, results,
+            arguments.kiosk_interval)))
+
+    await asyncio.sleep(arguments.seconds)
+    STOP = True
+    await asyncio.sleep(2.0)
+    for task in tasks:
+        task.cancel()
+
+    summary = {
+        'fleet': arguments.fleet,
+        'users': arguments.users,
+        'integrations': arguments.integrations,
+        'kiosks': arguments.kiosks,
+        'seconds': arguments.seconds,
+        'classes': {},
+    }
+    for bucket, entry in sorted(results.items()):
+        times = entry['times']
+        summary['classes'][bucket] = {
+            'samples': len(times),
+            'status': entry['status'],
+            'median_ms': round(statistics.median(times), 1) if times else 0.0,
+            'p95_ms': round(percentile(times, 0.95), 1),
+            'max_ms': round(max(times), 1) if times else 0.0,
+        }
+    print(json.dumps(summary, indent=2))
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

scripts/loadtest/poll_fleet.py

@@ -0,0 +1,141 @@
+"""Async load generator for the endpoint command long-poll channel.
+
+WHY this shape: the real agent (agent/ShopDbEndpointAgent.Core/CommandWorker.cs)
+re-polls immediately on a 204, so every agent holds a connection essentially
+continuously and fleet size IS the concurrency. This mirrors that loop rather
+than issuing independent requests, so the measured concurrency matches a real
+fleet of the same size.
+
+Runs INSIDE a container on the compose network. Generating from the Windows host
+would put host ephemeral-port and TLS-handshake limits in the measurement path,
+which would look like server saturation but would not be.
+
+While the fleet polls, a separate probe measures /api/dashboard/health - the
+user-visible UI latency - because that, not agent latency, is the symptom that
+matters for sizing.
+
+Stdlib asyncio only, so it needs no dependency install in the container.
+
+Usage:
+    python poll_fleet.py --agents 300 --seconds 90 --base https://api:5001
+"""
+import argparse
+import asyncio
+import json
+import os
+import ssl
+import statistics
+import time
+
+STOP = False
+
+
+def percentile(values, fraction):
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    index = min(int(len(ordered) * fraction), len(ordered) - 1)
+    return ordered[index]
+
+
+async def request(host, port, context, path, api_key, timeout):
+    """One HTTP/1.1 GET over a fresh connection. Returns (status, seconds)."""
+    started = time.monotonic()
+    writer = None
+    try:
+        reader, writer = await asyncio.wait_for(
+            asyncio.open_connection(host, port, ssl=context), timeout=timeout)
+        request_line = (
+            'GET %s HTTP/1.1\r\nHost: %s\r\nX-API-Key: %s\r\n'
+            'Accept: application/json\r\nConnection: close\r\n\r\n'
+            % (path, host, api_key))
+        writer.write(request_line.encode())
+        await writer.drain()
+        header = await asyncio.wait_for(reader.readuntil(b'\r\n'), timeout=timeout)
+        status = int(header.split()[1])
+        await asyncio.wait_for(reader.read(), timeout=timeout)
+        return status, time.monotonic() - started
+    except Exception as error:
+        return type(error).__name__, time.monotonic() - started
+    finally:
+        if writer is not None:
+            try:
+                writer.close()
+            except Exception:
+                pass
+
+
+async def agent_loop(index, host, port, context, api_key, results, timeout):
+    """Imitate one agent: poll, and on completion immediately poll again."""
+    hostname = 'LOADTEST%04d' % (index + 1)
+    path = '/api/endpoints/commands/next?hostname=' + hostname
+    while not STOP:
+        status, elapsed = await request(
+            host, port, context, path, api_key, timeout)
+        results.append((status, elapsed))
+        await asyncio.sleep(0)
+
+
+async def main():
+    global STOP
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--agents', type=int, required=True)
+    parser.add_argument('--seconds', type=int, default=90)
+    parser.add_argument('--host', default='api')
+    parser.add_argument('--port', type=int, default=5001)
+    parser.add_argument('--timeout', type=float, default=120.0)
+    parser.add_argument('--ramp', type=float, default=0.005)
+    arguments = parser.parse_args()
+
+    # A managed collector token, not the env key: the endpoints plugin has
+    # plugin.endpoints.envkeyfallback=false (the agent kill switch), so the env
+    # key no longer authorizes. See scripts/loadtest/loadtest_token.py.
+    api_key = os.environ.get('LOADTEST_TOKEN')
+    if not api_key:
+        raise SystemExit('LOADTEST_TOKEN not present in environment')
+
+    context = ssl.create_default_context()
+    context.check_hostname = False
+    context.verify_mode = ssl.CERT_NONE
+
+    results = []
+    tasks = []
+    started = time.monotonic()
+
+    for index in range(arguments.agents):
+        tasks.append(asyncio.create_task(agent_loop(
+            index, arguments.host, arguments.port, context, api_key,
+            results, arguments.timeout)))
+        if arguments.ramp:
+            await asyncio.sleep(arguments.ramp)
+
+    ramp_seconds = time.monotonic() - started
+
+    await asyncio.sleep(arguments.seconds)
+    STOP = True
+    for task in tasks:
+        task.cancel()
+    await asyncio.gather(*tasks, return_exceptions=True)
+
+    poll_ok = [d for s, d in results if s == 204]
+    poll_bad = [s for s, _ in results if s != 204]
+
+    failures = {}
+    for status in poll_bad:
+        failures[str(status)] = failures.get(str(status), 0) + 1
+
+    print(json.dumps({
+        'agents': arguments.agents,
+        'ramp_seconds': round(ramp_seconds, 1),
+        'duration_seconds': arguments.seconds,
+        'poll_completed': len(results),
+        'poll_ok_204': len(poll_ok),
+        'poll_failures': failures,
+        'poll_hold_median': round(statistics.median(poll_ok), 2) if poll_ok else None,
+        'poll_hold_p95': round(percentile(poll_ok, 0.95), 2) if poll_ok else None,
+        'poll_hold_max': round(max(poll_ok), 2) if poll_ok else None,
+    }, indent=2))
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

scripts/loadtest/pool_status.py

@@ -0,0 +1,42 @@
+"""Report SQLAlchemy pool saturation from inside the API container.
+
+WHY: MySQL's max_connections is NOT the ceiling that binds first. The effective
+limit is per-process: pool_size + max_overflow, multiplied by worker count.
+With the defaults (5 + 10) and 4 workers that is 60, not 151. A peak of 52
+therefore means 87 percent of the real ceiling, not 34 percent.
+
+Prints the computed ceiling so a measured high-water mark can be compared
+against the right number.
+
+Usage (inside the api container, PYTHONPATH=/app):
+    python pool_status.py
+"""
+import json
+import os
+
+from shopdb import create_app
+from shopdb.api import db
+
+
+def main():
+    app = create_app(os.environ.get('FLASK_ENV', 'production'))
+    with app.app_context():
+        engine = db.engine
+        pool = engine.pool
+        size = pool.size()
+        overflow = getattr(pool, '_max_overflow', 0)
+        workers = int(os.environ.get('GUNICORN_WORKERS', '4'))
+        print(json.dumps({
+            'pool_class': type(pool).__name__,
+            'pool_size': size,
+            'max_overflow': overflow,
+            'per_worker_ceiling': size + overflow,
+            'workers': workers,
+            'effective_ceiling': (size + overflow) * workers,
+            'checkedout': pool.checkedout(),
+            'status': pool.status(),
+        }, indent=2))
+
+
+if __name__ == '__main__':
+    main()

scripts/loadtest/sample_dbconns.py

@@ -0,0 +1,68 @@
+"""Sample live MySQL connection counts at high frequency.
+
+WHY: pool pressure from a long-poll fleet is BURSTY. The route releases its
+connection before sleeping and re-acquires on wake, so the interesting number is
+the peak concurrent checkout during a wake burst, not the steady-state count. A
+slow sampler misses it entirely.
+
+Runs on the HOST (needs the docker CLI).
+
+Usage:
+    python sample_dbconns.py --seconds 120 --interval 0.25
+"""
+import argparse
+import json
+import statistics
+import subprocess
+import time
+
+QUERY = ("SELECT COUNT(*), SUM(command<>'Sleep') "
+         "FROM information_schema.processlist WHERE user='shopdb';")
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--container', default='shopdb-flask-wilm-db-1')
+    parser.add_argument('--seconds', type=int, default=120)
+    parser.add_argument('--interval', type=float, default=0.25)
+    arguments = parser.parse_args()
+
+    total_samples = []
+    active_samples = []
+    deadline = time.monotonic() + arguments.seconds
+    command = ('mysql -u root -p$MYSQL_ROOT_PASSWORD -N -B -e "%s"' % QUERY)
+
+    while time.monotonic() < deadline:
+        try:
+            output = subprocess.check_output(
+                ['docker', 'exec', arguments.container, 'sh', '-c', command],
+                stderr=subprocess.DEVNULL, text=True).strip()
+        except subprocess.CalledProcessError:
+            break
+        if not output:
+            break
+        parts = output.split()
+        total_samples.append(int(parts[0]))
+        active_samples.append(int(parts[1]) if parts[1] != 'NULL' else 0)
+        time.sleep(arguments.interval)
+
+    def summarize(values):
+        if not values:
+            return {'samples': 0}
+        ordered = sorted(values)
+        index = min(int(len(ordered) * 0.95), len(ordered) - 1)
+        return {
+            'samples': len(values),
+            'median': statistics.median(values),
+            'p95': ordered[index],
+            'max': max(values),
+        }
+
+    print(json.dumps({
+        'connections': summarize(total_samples),
+        'active_queries': summarize(active_samples),
+    }, indent=2))
+
+
+if __name__ == '__main__':
+    main()

scripts/loadtest/sample_resources.py

@@ -0,0 +1,80 @@
+"""Sample container CPU and memory during a sizing run.
+
+WHY: the sizing table claims specific vCPU and RAM figures. Those can only be
+confirmed by watching the container under load, not by reasoning about greenlet
+size. This records peak RSS and CPU so a claimed "4 GB is enough" is either
+demonstrated or falsified.
+
+Runs on the HOST (needs the docker CLI), unlike the generators.
+
+Usage:
+    python sample_resources.py --container shopdb-sizing-api --seconds 120
+"""
+import argparse
+import json
+import re
+import statistics
+import subprocess
+import time
+
+
+def parse_size(text):
+    """Docker stats prints like '1.234GiB' / '512MiB'. Return megabytes."""
+    match = re.match(r'^([0-9.]+)\s*([KMGT]?i?B)$', text.strip())
+    if not match:
+        return 0.0
+    value = float(match.group(1))
+    unit = match.group(2).upper().replace('I', '')
+    factors = {'B': 1.0 / (1024 * 1024), 'KB': 1.0 / 1024, 'MB': 1.0,
+               'GB': 1024.0, 'TB': 1024.0 * 1024.0}
+    return value * factors.get(unit, 1.0)
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--container', required=True)
+    parser.add_argument('--seconds', type=int, default=120)
+    parser.add_argument('--interval', type=float, default=2.0)
+    arguments = parser.parse_args()
+
+    cpu_samples = []
+    memory_samples = []
+    deadline = time.monotonic() + arguments.seconds
+
+    while time.monotonic() < deadline:
+        try:
+            output = subprocess.check_output(
+                ['docker', 'stats', '--no-stream', '--format',
+                 '{{.CPUPerc}}|{{.MemUsage}}', arguments.container],
+                stderr=subprocess.DEVNULL, text=True).strip()
+        except subprocess.CalledProcessError:
+            break
+        if not output:
+            break
+        cpu_text, memory_text = output.split('|', 1)
+        cpu_samples.append(float(cpu_text.replace('%', '').strip()))
+        used = memory_text.split('/')[0]
+        memory_samples.append(parse_size(used))
+        time.sleep(arguments.interval)
+
+    def summarize(values):
+        if not values:
+            return {'samples': 0}
+        ordered = sorted(values)
+        index = min(int(len(ordered) * 0.95), len(ordered) - 1)
+        return {
+            'samples': len(values),
+            'median': round(statistics.median(values), 1),
+            'p95': round(ordered[index], 1),
+            'max': round(max(values), 1),
+        }
+
+    print(json.dumps({
+        'container': arguments.container,
+        'cpu_percent': summarize(cpu_samples),
+        'memory_mb': summarize(memory_samples),
+    }, indent=2))
+
+
+if __name__ == '__main__':
+    main()

scripts/loadtest/ui_probe.py

@@ -0,0 +1,78 @@
+"""Measure UI latency against the API, independent of any load generator.
+
+WHY separate from poll_fleet.py: when the probe shares an event loop with the
+simulated fleet, generator-side scheduling can starve the probe and produce UI
+latencies that are an artifact of the harness rather than of the server. Run
+this in its own container, concurrently with the fleet, so the two cannot
+interfere.
+
+Usage:
+    python ui_probe.py --seconds 90 --interval 1.0
+"""
+import argparse
+import json
+import ssl
+import statistics
+import time
+import urllib.request
+
+
+def percentile(values, fraction):
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    index = min(int(len(ordered) * fraction), len(ordered) - 1)
+    return ordered[index]
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--seconds', type=int, default=90)
+    parser.add_argument('--interval', type=float, default=1.0)
+    parser.add_argument('--url',
+                        default='https://api:5001/api/dashboard/health')
+    parser.add_argument('--timeout', type=float, default=120.0)
+    arguments = parser.parse_args()
+
+    context = ssl.create_default_context()
+    context.check_hostname = False
+    context.verify_mode = ssl.CERT_NONE
+
+    durations = []
+    failures = {}
+    deadline = time.monotonic() + arguments.seconds
+
+    while time.monotonic() < deadline:
+        started = time.monotonic()
+        try:
+            request = urllib.request.Request(arguments.url)
+            with urllib.request.urlopen(
+                    request, timeout=arguments.timeout,
+                    context=context) as response:
+                response.read()
+                if response.status == 200:
+                    durations.append(time.monotonic() - started)
+                else:
+                    key = str(response.status)
+                    failures[key] = failures.get(key, 0) + 1
+        except Exception as error:
+            key = type(error).__name__
+            failures[key] = failures.get(key, 0) + 1
+        remaining = arguments.interval - (time.monotonic() - started)
+        if remaining > 0:
+            time.sleep(remaining)
+
+    print(json.dumps({
+        'ui_samples': len(durations) + sum(failures.values()),
+        'ui_ok_200': len(durations),
+        'ui_failures': failures,
+        'ui_median_ms': round(statistics.median(durations) * 1000, 1)
+                        if durations else None,
+        'ui_p95_ms': round(percentile(durations, 0.95) * 1000, 1)
+                     if durations else None,
+        'ui_max_ms': round(max(durations) * 1000, 1) if durations else None,
+    }, indent=2))
+
+
+if __name__ == '__main__':
+    main()