fix(api): software report FK failure + BullMQ enqueue inside held DB context

apps/api/migrations/2026-07-17-a-device-vulnerabilities-software-fk-set-null.sql

@@ -0,0 +1,51 @@
+-- device_vulnerabilities.software_inventory_id had no ON DELETE action, so the
+-- agent software report (PUT /agents/:id/software), which wipes and reinserts a
+-- device's software_inventory rows, failed with FK violation 23503 for any
+-- device that had correlated software findings — freezing that device's
+-- software inventory permanently (Sentry BREEZE-3). SET NULL matches the
+-- correlation service's semantics: upsertDeviceVulnerability re-links the
+-- finding to the current inventory row on the next scan, and the report route
+-- re-links surviving software in the same transaction.
+--
+-- The constraint is added NOT VALID to skip the full-table validation scan
+-- while ALTER TABLE holds its lock on both tables (agent reports and
+-- correlation writes would queue behind it during deploy). Existing rows are
+-- guaranteed valid — the dropped constraint enforced the same reference — and
+-- the companion -b- migration VALIDATEs in a separate transaction under the
+-- weaker SHARE UPDATE EXCLUSIVE lock.
+DO $$
+BEGIN
+  IF EXISTS (
+    SELECT 1
+    FROM pg_constraint
+    WHERE conname = 'device_vulnerabilities_software_inventory_id_fkey'
+      AND conrelid = 'device_vulnerabilities'::regclass
+      AND confdeltype <> 'n'
+  ) THEN
+    ALTER TABLE device_vulnerabilities
+      DROP CONSTRAINT device_vulnerabilities_software_inventory_id_fkey;
+  END IF;
+
+  IF NOT EXISTS (
+    SELECT 1
+    FROM pg_constraint
+    WHERE conname = 'device_vulnerabilities_software_inventory_id_fkey'
+      AND conrelid = 'device_vulnerabilities'::regclass
+  ) THEN
+    ALTER TABLE device_vulnerabilities
+      ADD CONSTRAINT device_vulnerabilities_software_inventory_id_fkey
+      FOREIGN KEY (software_inventory_id) REFERENCES software_inventory(id)
+      ON DELETE SET NULL
+      NOT VALID;
+  END IF;
+END $$;
+
+-- The SET NULL referential trigger fires once per deleted software_inventory
+-- row (up to 10k per software report) and executes
+-- `UPDATE device_vulnerabilities SET software_inventory_id = NULL WHERE
+-- software_inventory_id = $1` — RI triggers bypass RLS, so without an index
+-- each firing is a scan of the whole multi-tenant table. The same index serves
+-- the report route's finding-snapshot join.
+CREATE INDEX IF NOT EXISTS device_vuln_software_inventory_idx
+  ON device_vulnerabilities (software_inventory_id)
+  WHERE software_inventory_id IS NOT NULL;

apps/api/migrations/2026-07-17-b-device-vulnerabilities-software-fk-validate.sql

@@ -0,0 +1,17 @@
+-- Companion to 2026-07-17-a: VALIDATE the NOT VALID FK in its own transaction.
+-- VALIDATE CONSTRAINT takes SHARE UPDATE EXCLUSIVE, so agent reports and
+-- correlation writes proceed during the scan (unlike a validating ADD
+-- CONSTRAINT, which would hold the ALTER's exclusive lock for the duration).
+DO $$
+BEGIN
+  IF EXISTS (
+    SELECT 1
+    FROM pg_constraint
+    WHERE conname = 'device_vulnerabilities_software_inventory_id_fkey'
+      AND conrelid = 'device_vulnerabilities'::regclass
+      AND NOT convalidated
+  ) THEN
+    ALTER TABLE device_vulnerabilities
+      VALIDATE CONSTRAINT device_vulnerabilities_software_inventory_id_fkey;
+  END IF;
+END $$;

apps/api/src/db/schema/vulnerabilityManagement.ts

@@ -127,7 +127,7 @@ export const deviceVulnerabilities = pgTable('device_vulnerabilities', {
   orgId: uuid('org_id').notNull().references(() => organizations.id),
   deviceId: uuid('device_id').notNull().references(() => devices.id),
   vulnerabilityId: uuid('vulnerability_id').notNull().references(() => vulnerabilities.id),
-  softwareInventoryId: uuid('software_inventory_id').references(() => softwareInventory.id),
+  softwareInventoryId: uuid('software_inventory_id').references(() => softwareInventory.id, { onDelete: 'set null' }),
   status: varchar('status', { length: 20 }).notNull().default('open'), // lifecycle: see VULN_STATUSES in @breeze/shared (source of truth: open|patched|mitigated|accepted)
   riskScore: numeric('risk_score', { precision: 5, scale: 2 }),         // CVSS-primary
   matchConfidence: varchar('match_confidence', { length: 16 }),
@@ -143,4 +143,7 @@ export const deviceVulnerabilities = pgTable('device_vulnerabilities', {
   orgDeviceIdx: index('device_vuln_org_device_idx').on(table.orgId, table.deviceId),
   statusIdx: index('device_vuln_status_idx').on(table.status),
   ticketIdx: index('device_vuln_ticket_id_idx').on(table.ticketId).where(sql`ticket_id IS NOT NULL`),
+  // Serves the ON DELETE SET NULL referential trigger, which fires per deleted
+  // software_inventory row (bypassing RLS) and would otherwise scan this table.
+  softwareInventoryIdx: index('device_vuln_software_inventory_idx').on(table.softwareInventoryId).where(sql`software_inventory_id IS NOT NULL`),
 }));

apps/api/src/routes/agentWs.enqueueContract.test.ts

@@ -0,0 +1,39 @@
+/**
+ * Static contract (#1105, BREEZE-H): every BullMQ enqueue call site in
+ * agentWs.ts runs under runOutsideDbContext. The whole command_result pipeline
+ * executes inside a held org-scoped transaction (runWithAgentDbAccess), so an
+ * unwrapped enqueue pins a pooled Postgres connection idle-in-transaction
+ * across Redis round-trips — and for instrumented queues fires the
+ * assertOutsideHeldDbContext tripwire straight into Sentry. Seven sites were
+ * fixed in one pass; this scan keeps the next one from regressing silently.
+ */
+import { describe, expect, it } from 'vitest';
+import { readFileSync } from 'node:fs';
+import path from 'node:path';
+
+const source = readFileSync(path.join(__dirname, 'agentWs.ts'), 'utf8');
+
+describe('agentWs enqueue context contract (#1105)', () => {
+  it('every enqueue* call site is wrapped in runOutsideDbContext', () => {
+    const lines = source.split('\n');
+    const callSites: number[] = [];
+    lines.forEach((line, idx) => {
+      if (!/\benqueue[A-Z]\w*\s*\(/.test(line)) return;
+      if (/^\s*import\b/.test(line) || /\bfrom '/.test(line) || /await import\(/.test(line)) return;
+      callSites.push(idx);
+    });
+
+    // Known sites: monitor, SNMP (orphaned + tracked), discovery x2, backup,
+    // DR reconcile. If the scan finds fewer, the regex rotted — fix the scan,
+    // don't delete the assertion.
+    expect(callSites.length).toBeGreaterThanOrEqual(7);
+
+    for (const idx of callSites) {
+      const window = lines.slice(Math.max(0, idx - 3), idx + 1).join('\n');
+      expect(
+        window.includes('runOutsideDbContext('),
+        `enqueue call at agentWs.ts:${idx + 1} must be wrapped in runOutsideDbContext (#1105):\n${lines[idx]}`
+      ).toBe(true);
+    }
+  });
+});

apps/api/src/routes/agentWs.test.ts

@@ -187,17 +187,19 @@ vi.mock('../services/backupResultPersistence', async (importOriginal) => {
   };
 });
 
-import { db } from '../db';
+import { db, runOutsideDbContext } from '../db';
 import { devices } from '../db/schema';
 import {
   createAgentWsHandlers,
   createAgentWsRoutes,
   validateAgentToken,
   disconnectAgent,
   isAgentConnected,
+  sendCommandToAgent,
   __resetCrossTenantDropsForTest,
   AGENT_WS_CAPABILITIES,
 } from './agentWs';
+import { isRedisAvailable } from '../services/redis';
 import { claimPendingCommandsForDevice } from '../services/commandDispatch';
 import { writeAuditEvent } from '../services/auditEvents';
 import { isAgentTenantActive } from '../services/tenantStatus';
@@ -770,6 +772,69 @@ describe('agent websocket command results', () => {
     expect(ws.send).toHaveBeenCalledWith(expect.stringContaining('"ack"'));
   });
 
+  // BREEZE-H: the command_result handler runs inside a held org-scoped
+  // transaction; BullMQ enqueues must be wrapped in runOutsideDbContext so
+  // the #1105 tripwire in bullmqQueue passes and nested DB work routes to
+  // the pool (the wrap does not release the outer transaction's connection —
+  // dispatching after the context closes is the deeper #1105 fix).
+  it('enqueues an accepted monitor result via runOutsideDbContext (#1105, BREEZE-H)', async () => {
+    const preValidatedAgent = { deviceId: 'device-123', orgId: 'org-123' };
+
+    // onOpen: register the socket so sendCommandToAgent records the
+    // orphaned-result expectation the monitor result is matched against.
+    vi.mocked(db.select).mockReturnValue(selectAgentDevice([]) as any);
+    vi.mocked(db.update).mockReturnValue(updateResult() as any);
+    const handlers = createAgentWsHandlers('agent-123', preValidatedAgent);
+    const ws = wsMock();
+    await handlers.onOpen({}, ws as any);
+
+    sendCommandToAgent('agent-123', {
+      id: 'mon-monitor-1-123',
+      type: 'network_ping',
+      payload: { monitorId: 'monitor-1' },
+    } as any);
+
+    vi.mocked(isRedisAvailable).mockReturnValue(true);
+    let outsideDepth = 0;
+    vi.mocked(runOutsideDbContext).mockImplementation((fn: any) => {
+      outsideDepth++;
+      try {
+        return fn();
+      } finally {
+        outsideDepth--;
+      }
+    });
+    // Record (not assert) inside the mock: the handler catches enqueue errors,
+    // so a failing expect() inside the implementation would be swallowed.
+    let enqueuedOutsideContext: boolean | null = null;
+    vi.mocked(enqueueMonitorCheckResult).mockImplementation(async () => {
+      enqueuedOutsideContext = outsideDepth > 0;
+      return 'job-1';
+    });
+
+    await handlers.onMessage({
+      data: JSON.stringify({
+        type: 'command_result',
+        commandId: 'mon-monitor-1-123',
+        status: 'completed',
+        result: {
+          monitorId: 'monitor-1',
+          status: 'online',
+          responseMs: 12
+        }
+      })
+    } as any, ws as any);
+
+    expect(enqueueMonitorCheckResult).toHaveBeenCalledTimes(1);
+    expect(enqueueMonitorCheckResult).toHaveBeenCalledWith(
+      'monitor-1',
+      expect.objectContaining({ monitorId: 'monitor-1', status: 'online', responseMs: 12 }),
+      expect.objectContaining({ source: 'route:agentWs:monitor-result' })
+    );
+    expect(enqueuedOutsideContext).toBe(true);
+    expect(ws.send).toHaveBeenCalledWith(expect.stringContaining('"ack"'));
+  });
+
   it('drops terminal output for sessions not owned by the connected agent', async () => {
     const preValidatedAgent = { deviceId: 'device-123', orgId: 'org-123' };
 

apps/api/src/routes/agentWs.ts

@@ -190,11 +190,14 @@ async function handleDiscoveryResult({ agentId, command, result }: Parameters<Co
         .limit(1);
 
       if (job && isRedisAvailable()) {
-        await enqueueDiscoveryResults(
+        const normalizedHosts = normalizeDiscoveryHosts(discoveryData.hosts);
+        // Exit the held org-scoped transaction context for the Redis
+        // round-trips (#1105) — see the note on the monitor-result branch.
+        await runOutsideDbContext(() => enqueueDiscoveryResults(
           expectedJobId,
           job.orgId,
           job.siteId,
-          normalizeDiscoveryHosts(discoveryData.hosts),
+          normalizedHosts,
           discoveryData.hostsScanned ?? 0,
           discoveryData.hostsDiscovered ?? 0,
           undefined,
@@ -204,7 +207,7 @@ async function handleDiscoveryResult({ agentId, command, result }: Parameters<Co
             actorId: agentId,
             source: 'route:agentWs:script-network-scan',
           }
-        );
+        ));
       } else if (job) {
         // Redis not available — mark job failed so user knows results weren't processed
         console.warn(`[AgentWs] Redis unavailable, cannot process ${discoveryData.hosts.length} discovery hosts for job ${expectedJobId}`);
@@ -369,7 +372,10 @@ async function handleSnmpPollResult({ agentId, command, result }: Parameters<Com
         return;
       }
       if (isRedisAvailable()) {
-        await enqueueSnmpPollResults(expectedDeviceId, snmpData.metrics);
+        const metrics = snmpData.metrics;
+        // Exit the held org-scoped transaction context for the Redis
+        // round-trips (#1105) — see the note on the monitor-result branch.
+        await runOutsideDbContext(() => enqueueSnmpPollResults(expectedDeviceId, metrics));
       } else {
         // Redis not available — log warning about dropped metrics and mark status
         console.warn(`[AgentWs] Redis unavailable, dropping ${snmpData.metrics.length} SNMP metrics for device ${expectedDeviceId}`);
@@ -382,6 +388,7 @@ async function handleSnmpPollResult({ agentId, command, result }: Parameters<Com
     }
   } catch (err) {
     console.error(`[AgentWs] Failed to process SNMP poll results for ${agentId}:`, err);
+    captureException(err);
   }
 }
 
@@ -1010,7 +1017,11 @@ export async function processOrphanedCommandResult(
     console.log(`[AgentWs] Processing SNMP poll result for device ${snmpData.deviceId} from agent ${agentId}`);
     try {
       if (isRedisAvailable()) {
-        await enqueueSnmpPollResults(snmpData.deviceId, snmpData.metrics, result.commandId);
+        // Exit the held org-scoped transaction context for the Redis
+        // round-trips (#1105) — see the note on the monitor-result branch.
+        await runOutsideDbContext(() =>
+          enqueueSnmpPollResults(snmpData.deviceId!, snmpData.metrics!, result.commandId)
+        );
       } else {
         console.warn(`[AgentWs] Redis unavailable, dropping ${snmpData.metrics.length} SNMP metrics for device ${snmpData.deviceId}`);
         const { snmpDevices } = await import('../db/schema');
@@ -1046,32 +1057,33 @@ export async function processOrphanedCommandResult(
     }
     console.log(`[AgentWs] Processing monitor check result for monitor ${monitorData.monitorId} from agent ${agentId}`);
     try {
-      const status = normalizeMonitorStatus(monitorData.status);
+      const monitorId = monitorData.monitorId;
+      const checkResult = {
+        monitorId,
+        checkId: result.commandId,
+        status: normalizeMonitorStatus(monitorData.status),
+        responseMs: monitorData.responseMs ?? 0,
+        statusCode: monitorData.statusCode,
+        error: monitorData.error,
+        details: monitorData as Record<string, unknown>
+      };
       if (isRedisAvailable()) {
-        await enqueueMonitorCheckResult(monitorData.monitorId, {
-          monitorId: monitorData.monitorId,
-          checkId: result.commandId,
-          status,
-          responseMs: monitorData.responseMs ?? 0,
-          statusCode: monitorData.statusCode,
-          error: monitorData.error,
-          details: monitorData as Record<string, unknown>
-        }, {
-          actorType: 'agent',
-          actorId: agentId,
-          source: 'route:agentWs:monitor-result',
-        });
+        // Command results are processed inside a held org-scoped transaction
+        // (runWithAgentDbAccess). runOutsideDbContext exits the ALS context so
+        // instrumented-queue tripwires pass and any nested DB work routes to
+        // the pool — it does NOT release the outer transaction's connection,
+        // which stays held for the (normally short) Redis round-trips; the
+        // full fix is dispatching enqueues after the context closes (#1105).
+        await runOutsideDbContext(() =>
+          enqueueMonitorCheckResult(monitorId, checkResult, {
+            actorType: 'agent',
+            actorId: agentId,
+            source: 'route:agentWs:monitor-result',
+          })
+        );
       } else {
-        console.warn(`[AgentWs] Redis unavailable, recording monitor result directly for ${monitorData.monitorId}`);
-        await recordMonitorCheckResult(monitorData.monitorId, {
-          monitorId: monitorData.monitorId,
-          checkId: result.commandId,
-          status,
-          responseMs: monitorData.responseMs ?? 0,
-          statusCode: monitorData.statusCode,
-          error: monitorData.error,
-          details: monitorData as Record<string, unknown>
-        });
+        console.warn(`[AgentWs] Redis unavailable, recording monitor result directly for ${monitorId}`);
+        await recordMonitorCheckResult(monitorId, checkResult);
       }
     } catch (err) {
       console.error(`[AgentWs] Failed to process monitor check result for ${agentId}:`, err);
@@ -1284,11 +1296,14 @@ export async function processOrphanedCommandResult(
       }
 
       if (isRedisAvailable()) {
-        await enqueueDiscoveryResults(
+        const normalizedHosts = normalizeDiscoveryHosts(discoveryData.hosts);
+        // Exit the held org-scoped transaction context for the Redis
+        // round-trips (#1105) — see the note on the monitor-result branch.
+        await runOutsideDbContext(() => enqueueDiscoveryResults(
           discoveryJob.id,
           discoveryJob.orgId,
           discoveryJob.siteId,
-          normalizeDiscoveryHosts(discoveryData.hosts),
+          normalizedHosts,
           discoveryData.hostsScanned ?? 0,
           discoveryData.hostsDiscovered ?? 0,
           undefined,
@@ -1298,7 +1313,7 @@ export async function processOrphanedCommandResult(
             actorId: agentId,
             source: 'route:agentWs:discovery-result',
           }
-        );
+        ));
       } else {
         console.warn(`[AgentWs] Redis unavailable, cannot process ${discoveryData.hosts.length} discovery hosts for job ${discoveryJob.id}`);
         await db
@@ -1422,7 +1437,9 @@ export async function processOrphanedCommandResult(
         : `Malformed backup result payload: ${parsedBackup.error.issues.map((issue) => issue.message).join(', ')}`;
 
       if (isRedisAvailable()) {
-        await enqueueBackupResults(
+        // Exit the held org-scoped transaction context for the Redis
+        // round-trips (#1105) — see the note on the monitor-result branch.
+        await runOutsideDbContext(() => enqueueBackupResults(
           backupJob.id,
           backupJob.orgId,
           backupJob.deviceId,
@@ -1449,7 +1466,7 @@ export async function processOrphanedCommandResult(
             actorId: agentId,
             source: 'route:agentWs:backup-result',
           }
-        );
+        ));
       } else {
         console.warn(`[AgentWs] Redis unavailable, marking backup job ${backupJob.id} with inline result`);
         const persisted = await applyBackupCommandResultToJob({
@@ -1741,7 +1758,10 @@ async function processCommandResult(
 
       try {
         const { enqueueDrExecutionReconcile } = await import('../jobs/drExecutionWorker');
-        await enqueueDrExecutionReconcile(commandPayload.drExecutionId);
+        const drExecutionId = commandPayload.drExecutionId as string;
+        // Exit the held org-scoped transaction context for the Redis
+        // round-trips (#1105) — see the note on the monitor-result branch.
+        await runOutsideDbContext(() => enqueueDrExecutionReconcile(drExecutionId));
       } catch (err) {
         console.error(`[AgentWs] Failed to enqueue DR reconciliation for ${result.commandId}:`, err);
         captureException(err);