ConnectWise Automate scoping & durable workflows

Status: draft for review · Author: engineering · Last updated: 2026-07-09

This document covers two related problems surfaced while integrating ConnectWise Automate:

1. Scoping — making sure we only ingest our devices from Automate, not every MSP client’s, including drilling into groups inside a client.
2. Durable workflows — making long-running syncs survive restarts instead of silently orphaning (as the Rapid7 sync did).

The ConnectWiseAutomateClient referenced throughout is documented in Backend & services; the in-process BackgroundTasks model discussed in Part 2 is what the durable worker in Docker & local setup would eventually own.

---

Part 1 — ConnectWise Automate scoping

1.1 How Automate is organized

lt.tfmx.com is Lockstep’s Automate server and it hosts 98 clients (companies). Automate’s hierarchy is:

Server
└── Client (company)          e.g. "Lockstep Technology Group, LLC" (Id 118)
    └── Location               e.g. "Staging", "Engineers"
        └── Computer (device)  e.g. LTG-62DTTJ4
    └── Groups                 cross-cutting device groups (by policy/OS/role)

• Client = a customer company. This is the primary tenancy boundary.
• Location = a sub-site within a client.
• Group = a saved device grouping (can span locations, driven by rules or manual membership). Groups are how Automate techs target policies/scripts.

1.2 What we do today (client scoping) — and it works

The integration scopes ingestion to one or more clients via the Automate condition filter:

webapp/backend/app/services/connectwise_automate_client.py

get_computers_for_client(client_id)  # -> GET /Computers?condition=Client.Id=<id>

Config (connectwise_automate_config) stores scope_client_ids / scope_client_names, editable from the CW Automate page’s Client scope selector. It is currently set to:

scope_client_ids:   ["118"]
scope_client_names: ["Lockstep Technology Group, LLC"]

Verified against the live API: Client.Id=118 returns 240 computers — all genuinely Lockstep’s internal machines. So client scoping is correct.

1.3 The incident that looked like a scoping failure

The Assets list showed devices like BC-LOAN (POEFCU\rmccoy), COBY (NEMSCO\ccrooks), BKDIG1 (AzureAD\BrianKing). Those are other clients’ machines:

Device	Real Automate client
BC-LOAN	POE Federal Credit Union (Id 5770)
COBY	National Electric Motor & Supply (Id 13627)
BKDIG1	Sun Theory (Id 20796)

Root cause: not a scoping bug — stale data. The very first Automate sync ran before scoping existed and pulled all 2,961 devices across every client into inventory_items. When we added scoping and re-synced, the connectwise_automate_devices collection was pruned to 240, but ~753 devices’ worth of software rows lingered in inventory_items, so they kept appearing as connectwise_automate-sourced assets.

Fix applied: purged inventory_items where source=connectwise_automate and the device isn’t in the scoped device set (753 → 243 devices; 25,266 → 8,692 rows), then rebuilt the software rollup.

Hardening (recommended): the per-sync prune already deletes inventory_items whose device_id ∉ live_ids. To make stale data structurally impossible, also stamp every CW row with client_id and prune by client_id ∉ scope_client_ids. That way a scope change alone (without a device-set diff) always evicts out-of-scope data on the next sync.

1.4 When client scoping isn’t enough → group scoping

Client scoping is sufficient today because Lockstep’s internal fleet is its own client (118 → 240 devices). But two situations call for group-level scoping:

• A client contains machines we don’t manage for vuln purposes (kiosks, lab boxes, decommissioned) and we only want a subset.
• We want to segment ingestion by role/site (servers vs workstations, per office) without creating separate clients.

Design:

1. Client → new methods (ConnectWiseAutomateClient):

   • get_groups() → GET /Groups (already partially present in the CLI port); returns Id, Name, FullName, ComputerCount, Parent.
   • get_computers_for_group(group_id) → GET /Computers?condition=Groups.Id=<id> (verify exact condition path against the API — Automate sometimes needs Groups.GroupId or a /Groups/{id}/Computers sub-resource).

2. Config (connectwise_automate_config): add optional scope_group_ids / scope_group_names. Resolution precedence:

   • if scope_group_ids set → fetch per group (most specific);
   • else if scope_client_ids set → fetch per client (current behavior);
   • else → all (discouraged; UI warns).

3. Sync (_run_sync): union the computers across the selected groups, dedupe by Id, then continue exactly as today (write devices + inventory_items(source=connectwise_automate, client_id, group_ids), prune stale, rebuild rollup).

4. UI (ConnectWiseAutomatePage): under the existing Client scope card, add a “Drill into groups” expander that lists the selected client’s groups (name + ComputerCount) with checkboxes and a search box — mirrors the client selector. Saving writes scope_group_ids; a sync then prunes to that set.

5. Prune: delete inventory_items where client_id ∈ scope_client_ids but group_ids ∩ scope_group_ids = ∅ (i.e. in-client but out-of-group), plus the existing device_id ∉ live_ids rule.

Effort: ~half a day (client method + condition verification, config fields, sync union, one UI card). Low risk — additive to the working client scope.

1.5 Recommendation

• Keep client scope = [118] as the default; it is correct and sufficient now.
• Add client_id stamping + client-based prune (small, prevents recurrence of the stale-data class of bug).
• Implement group scoping as an opt-in refinement when we need sub-client granularity. Not required today.

---

Part 2 — Durable workflows

2.1 The problem

Every integration sync (SentinelOne, Intune, Rapid7, ConnectWise Automate, Entra, CSV parse) is launched with FastAPI BackgroundTasks, i.e. it runs inside the web process:

background_tasks.add_task(_run_sync, tenant_id, db)   # in-process coroutine

Consequences:

• Orphaned on restart. When the API container restarts (deploy, crash, env change, the recent MongoDB disk recovery), any in-flight sync is killed, but sync_status stays "running" in Mongo forever. This is exactly what happened to Rapid7 (stuck at “batch 6/24” with a perpetual spinner).
• Not resumable. A killed sync restarts from scratch; the 22k findings it had collected are abandoned mid-stream.
• No visibility. No heartbeat, no “last progress at”, no history of past runs — only a single mutable status field on the config doc.
• Couples uptime to jobs. A heavy sync competes with request handling in the same process/threadpool.

We already applied a stale-sync guard to Rapid7 (sync-status flips running→error after 20 min of no updates) as a stopgap. That stops the spinner but doesn’t make syncs durable or resumable. This section proposes the real fix.

2.2 Requirements

1. Durable — a job survives an API restart; its state lives in the DB, not in process memory.
2. Self-healing — a job whose worker died is detected (missed heartbeat) and marked failed/requeued, never stuck “running”.
3. Observable — per-job status, progress, heartbeat, start/end, error, and a run history.
4. Idempotent & resumable — re-running is safe (upserts); ideally resumes from a checkpoint (e.g. last processed asset batch) instead of restarting.
5. Isolated — long jobs run off the request path so they don’t degrade the API.
6. Concurrency-safe — the same sync can’t run twice for one tenant at once.

2.3 What we already have

• A dedicated vulnmgmt-worker container running APScheduler (uv run -m vulnmgmt.worker.scheduler) — “lightweight alternative to Celery, no Redis required.” Today it runs scheduled jobs; it is the natural home for durable jobs too.
• MongoDB (already the system of record) — good enough to be the job store; we do not need to add Redis/Celery for our volume.

2.4 Proposed architecture — DB-backed job queue + worker

Introduce a sync_jobs collection as the durable unit of work, a small enqueue helper on the API, and a claim/execute loop in the worker.

sync_jobs document:

{
  "_id": ObjectId,
  "tenant_id": "…",
  "kind": "rapid7_sync" | "connectwise_automate_sync" | "entra_sync" | "csv_parse" | …,
  "args": { /* e.g. { "upload_id": "…" } */ },
  "status": "queued" | "running" | "success" | "error" | "canceled",
  "progress": { "phase": "findings", "done": 120, "total": 466, "detail": "…" },
  "checkpoint": { /* opaque, kind-specific resume token, e.g. { "batch": 6 } */ },
  "attempts": 0,
  "max_attempts": 3,
  "heartbeat_at": ISODate,   // worker updates every ~15s while running
  "lease_until": ISODate,    // claim expiry; another worker may steal after this
  "worker_id": "…",          // who owns it
  "error": null,
  "created_at": ISODate, "started_at": ISODate, "finished_at": ISODate
}

Enqueue (API side) — replaces background_tasks.add_task(...):

async def enqueue_job(db, tenant_id, kind, args=None):
    # de-dupe: don't queue if one is already active for this tenant+kind
    existing = await db.sync_jobs.find_one(
        {"tenant_id": tenant_id, "kind": kind, "status": {"$in": ["queued", "running"]}})
    if existing:
        return existing["_id"]
    return (await db.sync_jobs.insert_one({
        "tenant_id": tenant_id, "kind": kind, "args": args or {},
        "status": "queued", "attempts": 0, "max_attempts": 3,
        "created_at": utcnow(),
    })).inserted_id

Claim + execute (worker side) — atomic claim via find_one_and_update:

job = await db.sync_jobs.find_one_and_update(
    {"status": "queued"},                       # (or lease-expired running jobs)
    {"$set": {"status": "running", "worker_id": WID,
              "started_at": utcnow(),
              "heartbeat_at": utcnow(),
              "lease_until": utcnow() + 60s},
     "$inc": {"attempts": 1}},
    sort=[("created_at", 1)], return_document=AFTER)

The handler for each kind runs the existing _run_sync logic, but:

• updates progress + heartbeat_at + lease_until as it goes (every batch);
• writes a checkpoint after each batch so a re-run can $skip to it;
• on exception → status=error (or requeue if attempts < max_attempts).

Reaper (worker, every ~1 min): any running job whose lease_until has passed (worker died) → set back to queued (if attempts remain) or error. This is the durable, general version of the Rapid7 20-min guard — no job can ever be stuck.

API status endpoints read from sync_jobs instead of the mutable config.sync_status, giving real history + heartbeat for free.

2.5 Migration path (incremental, low-risk)

1. Phase 0 (done): stale-sync guard on Rapid7 so the current symptom is gone.
2. Phase 1: add sync_jobs + enqueue_job + a worker claim/execute loop for one integration (Rapid7 — the most painful). Keep the others on BackgroundTasks. Prove the model end-to-end.
3. Phase 2: add heartbeat + reaper + resume-from-checkpoint for Rapid7 (batch token). Now a restart mid-sync resumes instead of restarting.
4. Phase 3: migrate the remaining syncs (S1, Intune, CW Automate, Entra, CSV parse) to enqueue_job; delete the per-endpoint BackgroundTasks calls and the ad-hoc sync_status fields; point all sync-status endpoints at sync_jobs.
5. Phase 4 (optional): a small “Jobs” admin page (queued/running/failed, with retry/cancel) built on the same collection.

2.6 Why not Celery/RQ/Temporal?

• Celery/RQ need Redis and a broker; more moving parts than our volume warrants, and the team already chose APScheduler “to avoid Redis.”
• Temporal is the gold standard for durable workflows but is a heavy external service — overkill here.
• DB-backed queue on Mongo + the existing worker meets every requirement in §2.2 with infrastructure we already run. If throughput ever outgrows it, the sync_jobs abstraction is a clean seam to swap in a real broker later.

2.7 Effort

• Phase 1–2 (Rapid7 durable + resumable): ~1–1.5 days.
• Phase 3 (migrate all syncs): ~1 day.
• Phase 4 (Jobs UI): ~half a day.

---

Immediate action items

• Purge stale non-Lockstep CW Automate inventory_items (753 → 243 devices).
• Rapid7 stale-sync guard (20-min) so orphaned runs self-heal.
• Stamp CW rows with client_id + client-based prune (prevents stale-data recurrence).
• CW Automate group scoping (client method + config + sync union + UI card).
• Durable-workflow Phase 1–2 for Rapid7 (sync_jobs + worker + reaper + resume).
• Apply the 20-min stale guard to the other in-process syncs as a stopgap until Phase 3.