From 416434959a1fd1df67d21f99aa7f1223a94acc3e Mon Sep 17 00:00:00 2001 From: Billy Dunn Date: Fri, 7 Aug 2026 20:27:09 -0500 Subject: [PATCH] refactor(api): share the command-result handlers with the HTTP transport (#3097) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-command-type result handlers lived inside `agentWs.ts` and were only ever reachable over the websocket, so a result submitted to `POST /agents/:id/commands/:commandId/result` reached none of them. For `script` that means `script_executions` is never updated on the HTTP path. Measured on a live instance: 394 of 1070 executions (37%) carried a `timeout` status that did not match what happened, 89 of them having completed successfully with the output sitting in `device_commands.result` the whole time. #3172 stopped the reaper mislabelling those; this stops them being stranded. What moved ---------- The handler block and its `commandResultHandlers` registry move verbatim from `routes/agentWs.ts` to `services/commandResultHandlers.ts`. Everything they touch is an importable service or job, so there is no edge back into `agentWs`. The moved block is 385 lines before and after, and the diff between the two is eleven lines: - four handler signatures take `commandId` as a parameter, - the four `result.commandId` reads that fed them become that parameter, - two dynamic `./agents/helpers` imports become `../routes/agents/helpers`, - the registry gains `export`. Nothing else in the block changed. The two dynamic imports stay dynamic — they are dynamic to dodge a cycle. Why `commandId` is a parameter ------------------------------ The two transports disagree structurally, not just cosmetically: the websocket envelope carries `commandId` inline, while REST takes it from the path and authorizes against that path value. Reading an agent-supplied body id inside a handler would let it act on one command while ownership had been checked against another, so the id is supplied by whichever transport authorized it and is never read off the payload. Schema unification ------------------ The two `commandResultSchema` definitions diverged on the 1 MB cap: `routes/agents/schemas.ts` measured `Buffer.byteLength` while `agentWs.ts` used `.length` (UTF-16 code units), so the websocket accepted roughly 3x the intended budget for CJK-heavy output while REST rejected at 1 MB. The byte-accurate one is now canonical and `agentWs` extends it with the two envelope fields it adds (`type`, `commandId`). The cap is a `.refine()` on the result field rather than the object, so the shared base stays a plain `ZodObject` that REST consumes directly and the websocket can still `.extend()`. This is the one intentional behaviour change to the handlers themselves. Which types the HTTP path now dispatches ---------------------------------------- Five: `network_discovery`, `hyperv_backup`, `mssql_backup`, `snmp_poll` and `script` — the ones this route never post-processed at all. Deliberately not the whole registry. Thirteen of its eighteen keys already have an equivalent inline block in the route (backup verification, restore, vault sync, sensitive-data, CIS), and dispatching those as well would run each of them twice. Converging the overlapping thirteen is left as follow-up rather than folded in here: the CIS and sensitive-data handlers forward the derived stdout while the route's inline blocks forward `normalizedData` verbatim, so replacing them would change what those two post-processors receive. No DB-context wrap is added on the HTTP side. `agentAuthMiddleware` already holds an org-scoped `withDbAccessContext` open across the request with the same shape the websocket's `runWithAgentOrgDbAccess` builds; the websocket needs its own only because that path deliberately runs contextless (#3021). This route is not one of the `SELF_MANAGED_DB_CONTEXT_ACTIONS` — those match the final path segment, which here is `result`. A nested wrap would be a no-op regardless, and opening a second real transaction is the #1105 double-hold this route was cleaned up to avoid. Tests ----- - a script result over HTTP reaches the shared handler with the path-derived `commandId`, the resolved device and the derived stdout; - a type the route handles inline is NOT re-dispatched through the registry. Both were run against the un-fixed code first: emptying the dispatch set fails the first and only the first, and adding `cis_benchmark` to it fails the second. The #1105 enqueue contract scan now covers `services/commandResultHandlers.ts` as well as `agentWs.ts`. The pipeline's seven enqueue sites are now five plus two, and the contract is about the pipeline rather than the file — leaving the scan pinned to `agentWs.ts` would have silently dropped the two that moved. Confirmed by unwrapping one of the moved enqueues and watching the scan name it. --- .../routes/agentWs.enqueueContract.test.ts | 59 ++- apps/api/src/routes/agentWs.ts | 448 +---------------- apps/api/src/routes/agents/commands.test.ts | 81 ++++ apps/api/src/routes/agents/commands.ts | 76 +++ .../api/src/services/commandResultHandlers.ts | 459 ++++++++++++++++++ 5 files changed, 678 insertions(+), 445 deletions(-) create mode 100644 apps/api/src/services/commandResultHandlers.ts diff --git a/apps/api/src/routes/agentWs.enqueueContract.test.ts b/apps/api/src/routes/agentWs.enqueueContract.test.ts index 1c92020b4..90840d171 100644 --- a/apps/api/src/routes/agentWs.enqueueContract.test.ts +++ b/apps/api/src/routes/agentWs.enqueueContract.test.ts @@ -1,11 +1,19 @@ /** - * Static contract (#1105, BREEZE-H): every BullMQ enqueue call site in - * agentWs.ts runs under runOutsideDbContext. The whole command_result pipeline + * Static contract (#1105, BREEZE-H): every BullMQ enqueue call site in the + * command_result pipeline runs under runOutsideDbContext. That 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. + * + * #3097 split the pipeline across two files: the per-command-type handlers moved + * to services/commandResultHandlers.ts so the HTTP transport could dispatch them + * too, taking the discovery and SNMP enqueues with them. The contract is about + * the pipeline, not the file, so the scan covers both — otherwise moving a call + * site out of agentWs.ts would silently drop it from the guard, and the HTTP + * path runs these same handlers inside the request-long org context, where an + * unwrapped enqueue has exactly the same cost. */ import { describe, expect, it } from 'vitest'; import { readFileSync } from 'node:fs'; @@ -13,29 +21,42 @@ import path from 'node:path'; import { backupProcessResultSchema } from '../jobs/queueSchemas'; const source = readFileSync(path.join(__dirname, 'agentWs.ts'), 'utf8'); +const handlerSource = readFileSync( + path.join(__dirname, '..', 'services', 'commandResultHandlers.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); - }); + const scanned: Array<{ file: string; lines: string[] }> = [ + { file: 'agentWs.ts', lines: source.split('\n') }, + { file: 'services/commandResultHandlers.ts', lines: handlerSource.split('\n') }, + ]; - // 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); + let total = 0; + for (const { file, lines } of scanned) { + 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); + }); + total += callSites.length; - 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); + 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 ${file}:${idx + 1} must be wrapped in runOutsideDbContext (#1105):\n${lines[idx]}` + ).toBe(true); + } } + + // Known sites: monitor, SNMP (orphaned + tracked), discovery x2, backup, + // DR reconcile — five now in agentWs.ts and two in the extracted handlers. + // If the scan finds fewer, the regex rotted or a site moved to a third file + // — fix the scan, don't delete the assertion. + expect(total).toBeGreaterThanOrEqual(7); }); }); diff --git a/apps/api/src/routes/agentWs.ts b/apps/api/src/routes/agentWs.ts index fd72cecf9..994842105 100644 --- a/apps/api/src/routes/agentWs.ts +++ b/apps/api/src/routes/agentWs.ts @@ -62,6 +62,8 @@ import { SW_INSTALL_COMMAND_ID_REGEX, } from '../services/softwareDeploymentResult'; import { PG_UUID_REGEX, UUID_REGEX } from '../utils/uuid'; +import { commandResultSchema as baseCommandResultSchema } from './agents/schemas'; +import { commandResultHandlers, normalizeDiscoveryHosts } from '../services/commandResultHandlers'; /** Capabilities advertised to agents in the post-connect `connected` message. */ export const AGENT_WS_CAPABILITIES = ['terminal_output_base64', 'backup_run_async'] as const; @@ -153,412 +155,9 @@ function inferRestoreCommandType(restoreJob: { /** * Signature for per-command-type result handlers dispatched from processCommandResult. */ -type CommandResultHandler = (params: { - agentId: string; - command: typeof deviceCommands.$inferSelect; - result: z.infer; - resolvedDeviceId: string; - stdout: string | undefined; -}) => Promise; - -// --------------------------------------------------------------------------- -// Per-command-type result handlers (used by the dispatch map in processCommandResult) -// --------------------------------------------------------------------------- - -/** Coerce Date instances in host firstSeen/lastSeen to ISO strings so Zod datetime validation passes. */ -function normalizeDiscoveryHosts(hosts: DiscoveredHostResult[]): DiscoveredHostResult[] { - return hosts.map(h => ({ - ...h, - firstSeen: (h.firstSeen as any) instanceof Date ? (h.firstSeen as any).toISOString() : h.firstSeen, - lastSeen: (h.lastSeen as any) instanceof Date ? (h.lastSeen as any).toISOString() : h.lastSeen, - })); -} - -async function handleDiscoveryResult({ agentId, command, result }: Parameters[0]): Promise { - const payload = command.payload as Record | null; - const expectedJobId = typeof payload?.jobId === 'string' ? payload.jobId : null; - try { - const discoveryData = result.result as { - jobId?: string; - hosts?: DiscoveredHostResult[]; - hostsScanned?: number; - hostsDiscovered?: number; - adjacency?: DeviceAdjacency[]; - } | undefined; - - if (discoveryData?.hosts) { - if (!expectedJobId || discoveryData.jobId !== expectedJobId) { - console.warn( - `[AgentWs] Rejecting mismatched discovery result ${result.commandId} from agent ${agentId}: ` + - `sentJob=${discoveryData.jobId ?? 'none'} expected=${expectedJobId ?? 'none'}` - ); - return; - } - } - - if (expectedJobId && discoveryData?.hosts) { - // Look up the job to get orgId and siteId - const [job] = await db - .select({ orgId: discoveryJobs.orgId, siteId: discoveryJobs.siteId }) - .from(discoveryJobs) - .where(eq(discoveryJobs.id, expectedJobId)) - .limit(1); - - if (job && isRedisAvailable()) { - 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, - normalizedHosts, - discoveryData.hostsScanned ?? 0, - discoveryData.hostsDiscovered ?? 0, - undefined, - discoveryData.adjacency ?? [], - { - actorType: 'agent', - 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}`); - await db - .update(discoveryJobs) - .set({ - status: 'failed', - completedAt: new Date(), - hostsDiscovered: discoveryData.hostsDiscovered ?? 0, - hostsScanned: discoveryData.hostsScanned ?? 0, - errors: { message: 'Results received but could not be processed: job queue unavailable' }, - updatedAt: new Date() - }) - .where(eq(discoveryJobs.id, expectedJobId)); - } else { - console.warn( - `[AgentWs] Discovery job ${expectedJobId} not found in DB — ` + - `discarding ${discoveryData.hosts.length} host(s) from agent ${agentId}` - ); - } - } - } catch (err) { - console.error(`[AgentWs] Failed to process discovery results for ${agentId}:`, err); - captureException(err); - if (expectedJobId) { - try { - await db - .update(discoveryJobs) - .set({ - status: 'failed', - completedAt: new Date(), - errors: { message: err instanceof Error ? err.message : 'Failed to enqueue discovery results' }, - updatedAt: new Date() - }) - .where(eq(discoveryJobs.id, expectedJobId)); - } catch (dbErr) { - console.error(`[AgentWs] Additionally failed to mark discovery job ${expectedJobId} as failed:`, dbErr); - } - } - } -} - -async function handleBackupVerificationResult({ agentId, result, stdout }: Parameters[0]): Promise { - try { - await processBackupVerificationResult(result.commandId, { - status: result.status, - stdout, - error: result.error, - }); - } catch (err) { - console.error(`[AgentWs] Failed to process backup verification result for ${agentId}:`, err); - captureException(err); - } -} - -async function handleVmRestoreResult({ agentId, command, result, resolvedDeviceId }: Parameters[0]): Promise { - try { - await updateRestoreJobByCommandId({ - commandId: result.commandId, - deviceId: resolvedDeviceId, - commandType: command.type, - result, - }); - } catch (err) { - console.error(`[AgentWs] Failed to process queued restore result for ${agentId}:`, err); - captureException(err); - } -} - -async function handleProviderBackedBackupResult({ agentId, command, result, resolvedDeviceId }: Parameters[0]): Promise { - try { - const payload = - command.payload && typeof command.payload === 'object' && !Array.isArray(command.payload) - ? command.payload as Record - : {}; - const backupJobId = - typeof payload.backupJobId === 'string' - ? payload.backupJobId - : typeof payload.jobId === 'string' && UUID_REGEX.test(payload.jobId) - ? payload.jobId - : null; - - if (backupJobId) { - const [backupJob] = await db - .select({ - id: backupJobs.id, - orgId: backupJobs.orgId, - deviceId: backupJobs.deviceId, - }) - .from(backupJobs) - .where( - and( - eq(backupJobs.id, backupJobId), - eq(backupJobs.deviceId, resolvedDeviceId) - ) - ) - .limit(1); - - if (backupJob) { - const parsedBackup = backupCommandResultSchema.safeParse(result.result ?? {}); - if (!parsedBackup.success) { - await applyBackupCommandResultToJob({ - jobId: backupJob.id, - orgId: backupJob.orgId, - deviceId: backupJob.deviceId, - resultStatus: 'failed', - result: { - error: `Malformed backup result payload: ${parsedBackup.error.issues.map((issue) => issue.message).join(', ')}`, - }, - }); - } else { - await applyBackupCommandResultToJob({ - jobId: backupJob.id, - orgId: backupJob.orgId, - deviceId: backupJob.deviceId, - resultStatus: result.status, - // Provider-backed backups do not report `partial` today, but this - // path parses the agent's status and must not be the one place - // that silently discards it. - agentStatus: parsedBackup.data.status, - result: { - ...parsedBackup.data, - error: result.error || result.stderr, - }, - }); - } - } - } - } catch (err) { - console.error(`[AgentWs] Failed to process ${command.type} backup result for ${agentId}:`, err); - captureException(err); - } -} - -async function handleVaultSyncResult({ agentId, command, result, resolvedDeviceId, stdout }: Parameters[0]): Promise { - try { - await applyVaultSyncCommandResult({ - deviceId: resolvedDeviceId, - command, - resultStatus: result.status, - stdout, - stderr: result.stderr, - error: result.error, - }); - } catch (err) { - console.error(`[AgentWs] Failed to process vault sync result for ${agentId}:`, err); - captureException(err); - } -} - -async function handleSnmpPollResult({ agentId, command, result }: Parameters[0]): Promise { - try { - const payload = command.payload as Record | null; - const expectedDeviceId = typeof payload?.deviceId === 'string' ? payload.deviceId : null; - const snmpData = result.result as { - deviceId?: string; - metrics?: SnmpMetricResult[]; - } | undefined; - - if (snmpData?.deviceId && snmpData.metrics && snmpData.metrics.length > 0) { - if (!expectedDeviceId || snmpData.deviceId !== expectedDeviceId) { - console.warn( - `[AgentWs] Rejecting mismatched SNMP result ${result.commandId} from agent ${agentId}: ` + - `sentDevice=${snmpData.deviceId} expected=${expectedDeviceId ?? 'none'}` - ); - return; - } - if (isRedisAvailable()) { - 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}`); - const { snmpDevices } = await import('../db/schema'); - await db - .update(snmpDevices) - .set({ - lastPolled: new Date(), - // The device answered; only our own pipeline failed. Clear the - // failure backoff (#3217) so a Redis outage doesn't march every - // healthy SNMP target to 'offline' and a one-hour interval. - lastPollAttemptedAt: new Date(), - consecutiveFailures: 0, - lastStatus: 'warning' - }) - .where(eq(snmpDevices.id, expectedDeviceId)); - } - } - } catch (err) { - console.error(`[AgentWs] Failed to process SNMP poll results for ${agentId}:`, err); - captureException(err); - } -} - -async function handleScriptResult({ agentId, command, result, resolvedDeviceId, stdout }: Parameters[0]): Promise { - try { - const payload = command.payload as Record | null; - const executionId = payload?.executionId as string | undefined; - // #3162: `script_executions.id` is a uuid column, so a non-uuid - // executionId makes the UPDATE below throw with `invalid input syntax for - // type uuid` — swallowed by the catch at the bottom of this function, - // taking the agent's stdout with it. - // - // Nothing should mint a non-uuid executionId any more (the automation - // `execute_command` action, the only producer, now omits the field - // entirely). This guard is for commands queued BEFORE that deploy and still - // in flight, so it reports rather than silently skipping: a fresh non-uuid - // id means an unknown producer is sending garbage. - if (executionId && !PG_UUID_REGEX.test(executionId)) { - console.warn( - `[AgentWs] Skipping script_executions update for non-uuid executionId ${executionId} (command ${command.id})` - ); - captureException( - new Error('Non-uuid executionId in script command payload'), - undefined, - { commandId: command.id, agentId, executionId }, - ); - return; - } - if (executionId) { - let scriptStatus: 'completed' | 'failed' | 'timeout'; - if (result.status === 'completed') { - scriptStatus = result.exitCode && result.exitCode !== 0 ? 'failed' : 'completed'; - } else if (result.status === 'timeout') { - scriptStatus = 'timeout'; - } else { - scriptStatus = 'failed'; - } - - const updatedExecutions = await db - .update(scriptExecutions) - .set({ - status: scriptStatus, - completedAt: new Date(), - exitCode: result.exitCode ?? null, - // #2434: script output/errors surface to scripts:read users in the - // web UI — redact secrets before persistence (idempotent when the - // ingest chokepoint already redacted error/stderr). - stdout: stdout != null ? redactSecretsFromOutput(stdout) : null, - stderr: redactOptionalSecretText(result.stderr) ?? null, - errorMessage: redactOptionalSecretText(result.error) ?? null, - }) - .where(and( - eq(scriptExecutions.id, executionId), - eq(scriptExecutions.deviceId, resolvedDeviceId), - inArray(scriptExecutions.status, ['pending', 'queued', 'running']) - )) - .returning({ - id: scriptExecutions.id, - scriptId: scriptExecutions.scriptId, - }); - - // Update batch counters if this is part of a batch - const batchId = payload?.batchId as string | undefined; - if (batchId && updatedExecutions[0]) { - const counterField = scriptStatus === 'completed' ? 'devicesCompleted' : 'devicesFailed'; - await db - .update(scriptExecutionBatches) - .set({ - [counterField]: sql`${scriptExecutionBatches[counterField]} + 1` - }) - .where(and( - eq(scriptExecutionBatches.id, batchId), - eq(scriptExecutionBatches.scriptId, updatedExecutions[0].scriptId) - )); - } - } - } catch (err) { - // #3162 lived undetected because this catch logged to the container and - // nothing else — a swallowed 22P02 silently discarded every automation's - // script output. Report it like the SNMP handler does so the next failure - // in here (schema drift, a redaction throw, a batch-counter FK violation) - // surfaces instead of quietly eating results. - console.error(`[AgentWs] Failed to process script result for ${agentId}:`, err); - captureException(err, undefined, { - commandId: command.id, - agentId, - executionId: String((command.payload as Record | null)?.executionId ?? ''), - }); - } -} - -async function handleSensitiveDataResult({ agentId, command, result, stdout }: Parameters[0]): Promise { - try { - const { handleSensitiveDataCommandResult } = await import('./agents/helpers'); - await handleSensitiveDataCommandResult(command, { - status: result.status, - exitCode: result.exitCode, - stdout, - stderr: result.stderr, - durationMs: result.durationMs, - error: result.error, - } as any); - } catch (err) { - console.error(`[AgentWs] Failed to process sensitive data result for ${agentId}:`, err); - } -} - -async function handleCisResult({ agentId, command, result, stdout }: Parameters[0]): Promise { - try { - const { handleCisCommandResult } = await import('./agents/helpers'); - await handleCisCommandResult(command, { - status: result.status, - exitCode: result.exitCode, - stdout, - stderr: result.stderr, - durationMs: result.durationMs, - error: result.error, - } as any); - } catch (err) { - console.error(`[AgentWs] Failed to process CIS result for ${agentId}:`, err); - } -} - -const commandResultHandlers: Record = { - network_discovery: handleDiscoveryResult, - backup_verify: handleBackupVerificationResult, - backup_test_restore: handleBackupVerificationResult, - backup_restore: handleVmRestoreResult, - vm_restore_from_backup: handleVmRestoreResult, - vm_instant_boot: handleVmRestoreResult, - bmr_recover: handleVmRestoreResult, - hyperv_backup: handleProviderBackedBackupResult, - mssql_backup: handleProviderBackedBackupResult, - vault_sync: handleVaultSyncResult, - snmp_poll: handleSnmpPollResult, - script: handleScriptResult, - sensitive_data_scan: handleSensitiveDataResult, - encrypt_file: handleSensitiveDataResult, - secure_delete_file: handleSensitiveDataResult, - quarantine_file: handleSensitiveDataResult, - cis_benchmark: handleCisResult, - apply_cis_remediation: handleCisResult, -}; +// #3097: the command-result handlers and their registry now live in +// `services/commandResultHandlers.ts` so the HTTP transport can dispatch the +// same code. See that module for the handler bodies. // IMPORTANT #1 (#2556): when a verify/restore result is REJECTED by validation // (malformed payload deepJsonParse can't rescue, or oversize stdout tripping the @@ -706,26 +305,23 @@ function consumeOrphanedResultExpectation(agentId: string, commandId: string): O } // Message types from agent -const commandResultSchema = z.object({ +// #3097: one definition of the agent command-result payload for BOTH transports. +// +// The shared base lives in `routes/agents/schemas.ts`; the websocket envelope is +// that base plus the two fields only a socket needs — `type` (discriminator) and +// `commandId` (no URL path to read it from). REST consumes the base directly and +// takes its id from the path. +// +// This also retires a real divergence: the copy that used to live here measured +// the 1 MB `result` cap with `.length` (UTF-16 code units), so this path accepted +// roughly 3x the intended budget for CJK-heavy output while REST rejected at +// 1 MB. The base's cap is a `.refine()` on the `result` FIELD rather than on the +// object, which is what makes `.extend()` possible here — a `.refine()` on the +// object would return a `ZodEffects` with no `.extend()`, and we would be back to +// duplicating the very thing this removes. +const commandResultSchema = baseCommandResultSchema.extend({ type: z.literal('command_result'), commandId: z.string(), - status: z.enum(['completed', 'failed', 'timeout']), - exitCode: z.number().int().optional(), - stdout: z.string().max(5_000_000).optional(), - stderr: z.string().max(5_000_000).optional(), - durationMs: z.number().int().optional(), - // RFC3339 timestamp captured by the agent at the moment the command's - // primary work began. Optional for back-compat with pre-startedAt agents, - // which the server falls back to reconstructing from durationMs. - startedAt: z.string().datetime().optional(), - error: z.string().max(10_000).optional(), - result: z.any().optional().refine( - (val) => { - if (val === undefined || val === null) return true; - try { return JSON.stringify(val).length <= 1_048_576; } catch { return false; } - }, - { message: 'Command result payload exceeds 1 MB limit' } - ) }); type AgentCommandResult = z.infer; @@ -2060,7 +1656,7 @@ async function processCommandResult( // Short org wrap (#3021): handlers touch RLS-guarded org tables // through the ambient db (same as the happy-path dispatch below). await runWithAgentOrgDbAccess('agentWs.commandResult.handler', orgId, () => - rejectedHandler({ agentId, command, result: normalizedResult, resolvedDeviceId: resolvedDeviceId!, stdout }) + rejectedHandler({ agentId, command, commandId: result.commandId, result: normalizedResult, resolvedDeviceId: resolvedDeviceId!, stdout }) ); } catch (handlerErr) { console.error(`[AgentWs] Failed to finalize rejected ${command.type} result ${result.commandId}:`, handlerErr); @@ -2117,7 +1713,7 @@ async function processCommandResult( const handler = commandResultHandlers[command.type]; if (handler) { await runWithAgentOrgDbAccess('agentWs.commandResult.handler', orgId, () => - handler({ agentId, command, result: normalizedResult, resolvedDeviceId: resolvedDeviceId!, stdout }) + handler({ agentId, command, commandId: result.commandId, result: normalizedResult, resolvedDeviceId: resolvedDeviceId!, stdout }) ); } } catch (error) { diff --git a/apps/api/src/routes/agents/commands.test.ts b/apps/api/src/routes/agents/commands.test.ts index 0f91d61d6..f5f109d30 100644 --- a/apps/api/src/routes/agents/commands.test.ts +++ b/apps/api/src/routes/agents/commands.test.ts @@ -79,6 +79,20 @@ vi.mock('../../services/auditBaselineService', () => ({ processCollectedAuditPolicyCommandResult: vi.fn(), })); +// #3097 — the shared registry this route now dispatches for the handful of +// types it never post-processed inline. `cis_benchmark` is present here on +// purpose: it is one of the thirteen keys this route DOES handle inline, so its +// mock existing proves the route skips the registry by intent rather than +// because the handler happened to be missing. +const scriptRegistryHandlerMock = vi.fn().mockResolvedValue(undefined); +const cisRegistryHandlerMock = vi.fn().mockResolvedValue(undefined); +vi.mock('../../services/commandResultHandlers', () => ({ + commandResultHandlers: { + script: (...args: unknown[]) => scriptRegistryHandlerMock(...(args as [])), + cis_benchmark: (...args: unknown[]) => cisRegistryHandlerMock(...(args as [])), + }, +})); + vi.mock('../../services/sentry', () => ({ captureException: vi.fn(), // BREEZE-X: the terminal CAS now runs through dbWriteExpectingRows, which @@ -88,6 +102,7 @@ vi.mock('../../services/sentry', () => ({ })); import { captureMessage } from '../../services/sentry'; +import { handleCisCommandResult } from './helpers'; import { commandsRoutes } from './commands'; import { and, eq } from 'drizzle-orm'; import { deploymentResults } from '../../db/schema'; @@ -164,6 +179,72 @@ describe('agent commands routes', () => { } ); + // #3097 — a script result submitted over HTTP used to reach no per-type + // handler at all, so script_executions kept whatever status the reaper had + // written while the real output sat in device_commands.result. + it('dispatches a script result to the shared handler over the HTTP path', async () => { + const command = { + id: commandId, + deviceId: 'device-1', + type: 'script', + status: 'sent', + payload: { executionId: '33333333-3333-4333-8333-333333333333' }, + }; + selectMock.mockReturnValueOnce(chainMock([command])); + updateMock.mockReturnValueOnce(chainMock([{ id: 'cmd-1' }])); + + const res = await app.request(`/agents/${agentId}/commands/${commandId}/result`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + commandId, + status: 'completed', + exitCode: 0, + stdout: 'hello from the script', + }), + }); + + expect(res.status).toBe(200); + expect(scriptRegistryHandlerMock).toHaveBeenCalledTimes(1); + expect(scriptRegistryHandlerMock).toHaveBeenCalledWith({ + agentId: 'agent-1', + command: expect.objectContaining({ id: commandId, type: 'script' }), + // The id comes from the authorized path param, never the request body. + commandId, + result: expect.objectContaining({ status: 'completed', exitCode: 0 }), + resolvedDeviceId: 'device-1', + stdout: 'hello from the script', + }); + }); + + // The other half of the contract: types this route already post-processes + // inline must NOT also go through the registry, or every one of them would + // run twice. + it('does not re-dispatch a type it already handles inline', async () => { + selectMock.mockReturnValueOnce( + chainMock([ + { + id: commandId, + deviceId: 'device-1', + type: 'cis_benchmark', + status: 'sent', + }, + ]) + ); + updateMock.mockReturnValueOnce(chainMock([{ id: 'cmd-1' }])); + + const res = await app.request(`/agents/${agentId}/commands/${commandId}/result`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ commandId, status: 'completed', stdout: '{}' }), + }); + + expect(res.status).toBe(200); + expect(cisRegistryHandlerMock).not.toHaveBeenCalled(); + // …and the inline path is what actually ran. + expect(vi.mocked(handleCisCommandResult)).toHaveBeenCalledTimes(1); + }); + it('claims commands for the authenticated credential role only', async () => { claimPendingCommandsForDeviceMock.mockResolvedValueOnce([ { diff --git a/apps/api/src/routes/agents/commands.ts b/apps/api/src/routes/agents/commands.ts index 6dad6df43..8b1686c75 100644 --- a/apps/api/src/routes/agents/commands.ts +++ b/apps/api/src/routes/agents/commands.ts @@ -42,6 +42,39 @@ import { export const commandsRoutes = new Hono(); const ACCEPTED_COMMAND_RESULT_STATUSES = ['pending', 'sent'] as const; +/** + * #3097 — registry-backed command types this route dispatches to the shared + * handlers in `services/commandResultHandlers.ts`. + * + * Those handlers used to live inside `agentWs.ts` and were only ever reachable + * over the websocket, so a result submitted over HTTP silently skipped them. + * For `script` that meant `script_executions` was never updated: measured on a + * live instance, 394 of 1070 executions (37%) carried a `timeout` status that + * did not match reality, 89 of them having completed successfully with output + * sitting in `device_commands.result`. + * + * This set is deliberately NOT the whole registry. Thirteen of its eighteen + * keys already have an equivalent inline block further down this handler + * (backup_verify, backup_test_restore, backup_restore, bmr_recover, + * vm_restore_from_backup, vm_instant_boot, vault_sync, sensitive_data_scan, + * encrypt_file, secure_delete_file, quarantine_file, cis_benchmark, + * apply_cis_remediation) — dispatching those here as well would run each of + * them twice. Only the five the HTTP path never handled at all are listed. + * + * Converging the overlapping thirteen onto the registry is left as follow-up + * rather than folded in here: the CIS and sensitive-data handlers forward the + * *derived* stdout while this route's inline blocks forward `normalizedData` + * verbatim, so replacing them would change what those two post-processors + * receive — a behaviour change beyond this PR's one intentional one. + */ +const REGISTRY_DISPATCHED_COMMAND_TYPES = new Set([ + 'network_discovery', + 'hyperv_backup', + 'mssql_backup', + 'snmp_poll', + 'script', +]); + function commandResultToStdout(data: z.infer): string | undefined { return data.stdout ?? (data.result !== undefined ? JSON.stringify(data.result) : undefined); @@ -527,6 +560,49 @@ commandsRoutes.post( } } + // #3097 — the shared per-type handlers this transport never registered. + // + // No DB-context wrap here, unlike the websocket twin's + // `runWithAgentOrgDbAccess`: agentAuthMiddleware already holds an + // org-scoped `withDbAccessContext` open around this whole request, with the + // same shape the websocket wrap builds (scope 'organization', the device's + // orgId, accessibleOrgIds [orgId], no partner access). The websocket needs + // its own wrap only because that path deliberately runs contextless (#3021). + // This route is not one of the SELF_MANAGED_DB_CONTEXT_ACTIONS — those match + // on the final path segment, which here is `result`, not `commands` — so the + // request-long wrap is active. Adding a nested one would be a no-op anyway: + // `withDbAccessContext` returns `fn()` unchanged when a context is already + // on the async-local store, and opening a second real transaction is the + // #1105 double-hold this route was explicitly cleaned up to avoid. + if (REGISTRY_DISPATCHED_COMMAND_TYPES.has(command.type)) { + // Imported dynamically, like the DR handler below: the registry pulls in + // the discovery and SNMP workers, and through them the Drizzle schema + // module, which is more than this hot route should carry in its static + // graph — and enough to break suites that partially mock `db/schema`. + const { commandResultHandlers } = await import('../../services/commandResultHandlers'); + const handler = commandResultHandlers[command.type]; + if (handler) { + try { + await handler({ + // Handlers use this for log lines and one audit `actorId`, never a + // lookup. Prefer the authenticated agent record over the path + // param, matching this route's own writeAuditEvent actor below. + agentId: agent.agentId ?? agentId, + command, + commandId, + result: normalizedData, + // The lookup above constrains deviceId to the authenticated agent's + // device, so these are the same value the websocket resolves. + resolvedDeviceId: command.deviceId, + stdout, + }); + } catch (err) { + console.error(`[agents] shared ${command.type} result handler failed for ${commandId}:`, err); + captureException(err); + } + } + } + writeAuditEvent(c, { orgId: agent?.orgId, actorType: 'agent', diff --git a/apps/api/src/services/commandResultHandlers.ts b/apps/api/src/services/commandResultHandlers.ts new file mode 100644 index 000000000..98ec35810 --- /dev/null +++ b/apps/api/src/services/commandResultHandlers.ts @@ -0,0 +1,459 @@ +/** + * Shared agent command-result handlers (#3097). + * + * These handlers were defined inside `routes/agentWs.ts`, where only the + * WebSocket transport could reach them. Results submitted over the HTTP path + * (`routes/agents/commands.ts`) therefore never ran them at all — most visibly, + * `script` results never reached `script_executions`, leaving rows pending until + * the stale reaper stamped them `timeout`. + * + * Nothing here is new: the handler bodies below are the ones that were in + * `agentWs.ts`, moved verbatim so both transports dispatch the same code. The + * module boundary is the only thing that changed. + */ + +import { z } from 'zod'; +import { eq, and, inArray, sql } from 'drizzle-orm'; +import { db, runOutsideDbContext } from '../db'; +import { + deviceCommands, + discoveryJobs, + scriptExecutions, + scriptExecutionBatches, + backupJobs, +} from '../db/schema'; +import { enqueueDiscoveryResults, type DiscoveredHostResult, type DeviceAdjacency } from '../jobs/discoveryWorker'; +import { enqueueSnmpPollResults, type SnmpMetricResult } from '../jobs/snmpWorker'; +import { isRedisAvailable } from './redis'; +import { processBackupVerificationResult } from '../routes/backup/verificationService'; +import { applyBackupCommandResultToJob } from './backupResultPersistence'; +import { applyVaultSyncCommandResult } from './vaultSyncPersistence'; +import { backupCommandResultSchema } from '../routes/backup/resultSchemas'; +import { redactSecretsFromOutput, redactOptionalSecretText } from './secretRedaction'; +import { updateRestoreJobByCommandId } from './restoreResultPersistence'; +import { captureException } from './sentry'; +import { PG_UUID_REGEX, UUID_REGEX } from '../utils/uuid'; +// #3097: one definition of the agent command-result shape for BOTH transports. +// `schemas.ts` measures the 1 MB cap with `Buffer.byteLength` (bytes); the copy +// that used to live in `agentWs.ts` used `.length` (UTF-16 code units), so the +// websocket path accepted roughly 3x the intended budget for CJK-heavy output +// while the REST path rejected at 1 MB. The byte-accurate one wins. +import { commandResultSchema } from '../routes/agents/schemas'; + +export type CommandResultHandler = (params: { + agentId: string; + command: typeof deviceCommands.$inferSelect; + /** + * #3097: supplied by the transport, never read off the payload. + * + * The websocket envelope carries `commandId` inline; the REST route takes it + * from the path (`/:id/commands/:commandId/result`) and authorizes against + * that path value. Accepting an agent-supplied id in the body would let a + * handler act on one command while ownership was checked against another — + * so there is exactly one id here, and the transport that authorized it is + * the transport that passes it. + */ + commandId: string; + result: z.infer; + resolvedDeviceId: string; + stdout: string | undefined; +}) => Promise; + +// --------------------------------------------------------------------------- +// Per-command-type result handlers (used by the dispatch map in processCommandResult) +// --------------------------------------------------------------------------- + +/** Coerce Date instances in host firstSeen/lastSeen to ISO strings so Zod datetime validation passes. */ +export function normalizeDiscoveryHosts(hosts: DiscoveredHostResult[]): DiscoveredHostResult[] { + return hosts.map(h => ({ + ...h, + firstSeen: (h.firstSeen as any) instanceof Date ? (h.firstSeen as any).toISOString() : h.firstSeen, + lastSeen: (h.lastSeen as any) instanceof Date ? (h.lastSeen as any).toISOString() : h.lastSeen, + })); +} + +async function handleDiscoveryResult({ agentId, command, result, commandId }: Parameters[0]): Promise { + const payload = command.payload as Record | null; + const expectedJobId = typeof payload?.jobId === 'string' ? payload.jobId : null; + try { + const discoveryData = result.result as { + jobId?: string; + hosts?: DiscoveredHostResult[]; + hostsScanned?: number; + hostsDiscovered?: number; + adjacency?: DeviceAdjacency[]; + } | undefined; + + if (discoveryData?.hosts) { + if (!expectedJobId || discoveryData.jobId !== expectedJobId) { + console.warn( + `[AgentWs] Rejecting mismatched discovery result ${commandId} from agent ${agentId}: ` + + `sentJob=${discoveryData.jobId ?? 'none'} expected=${expectedJobId ?? 'none'}` + ); + return; + } + } + + if (expectedJobId && discoveryData?.hosts) { + // Look up the job to get orgId and siteId + const [job] = await db + .select({ orgId: discoveryJobs.orgId, siteId: discoveryJobs.siteId }) + .from(discoveryJobs) + .where(eq(discoveryJobs.id, expectedJobId)) + .limit(1); + + if (job && isRedisAvailable()) { + 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, + normalizedHosts, + discoveryData.hostsScanned ?? 0, + discoveryData.hostsDiscovered ?? 0, + undefined, + discoveryData.adjacency ?? [], + { + actorType: 'agent', + 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}`); + await db + .update(discoveryJobs) + .set({ + status: 'failed', + completedAt: new Date(), + hostsDiscovered: discoveryData.hostsDiscovered ?? 0, + hostsScanned: discoveryData.hostsScanned ?? 0, + errors: { message: 'Results received but could not be processed: job queue unavailable' }, + updatedAt: new Date() + }) + .where(eq(discoveryJobs.id, expectedJobId)); + } else { + console.warn( + `[AgentWs] Discovery job ${expectedJobId} not found in DB — ` + + `discarding ${discoveryData.hosts.length} host(s) from agent ${agentId}` + ); + } + } + } catch (err) { + console.error(`[AgentWs] Failed to process discovery results for ${agentId}:`, err); + captureException(err); + if (expectedJobId) { + try { + await db + .update(discoveryJobs) + .set({ + status: 'failed', + completedAt: new Date(), + errors: { message: err instanceof Error ? err.message : 'Failed to enqueue discovery results' }, + updatedAt: new Date() + }) + .where(eq(discoveryJobs.id, expectedJobId)); + } catch (dbErr) { + console.error(`[AgentWs] Additionally failed to mark discovery job ${expectedJobId} as failed:`, dbErr); + } + } + } +} + +async function handleBackupVerificationResult({ agentId, result, stdout, commandId }: Parameters[0]): Promise { + try { + await processBackupVerificationResult(commandId, { + status: result.status, + stdout, + error: result.error, + }); + } catch (err) { + console.error(`[AgentWs] Failed to process backup verification result for ${agentId}:`, err); + captureException(err); + } +} + +async function handleVmRestoreResult({ agentId, command, result, resolvedDeviceId, commandId }: Parameters[0]): Promise { + try { + await updateRestoreJobByCommandId({ + commandId: commandId, + deviceId: resolvedDeviceId, + commandType: command.type, + result, + }); + } catch (err) { + console.error(`[AgentWs] Failed to process queued restore result for ${agentId}:`, err); + captureException(err); + } +} + +async function handleProviderBackedBackupResult({ agentId, command, result, resolvedDeviceId }: Parameters[0]): Promise { + try { + const payload = + command.payload && typeof command.payload === 'object' && !Array.isArray(command.payload) + ? command.payload as Record + : {}; + const backupJobId = + typeof payload.backupJobId === 'string' + ? payload.backupJobId + : typeof payload.jobId === 'string' && UUID_REGEX.test(payload.jobId) + ? payload.jobId + : null; + + if (backupJobId) { + const [backupJob] = await db + .select({ + id: backupJobs.id, + orgId: backupJobs.orgId, + deviceId: backupJobs.deviceId, + }) + .from(backupJobs) + .where( + and( + eq(backupJobs.id, backupJobId), + eq(backupJobs.deviceId, resolvedDeviceId) + ) + ) + .limit(1); + + if (backupJob) { + const parsedBackup = backupCommandResultSchema.safeParse(result.result ?? {}); + if (!parsedBackup.success) { + await applyBackupCommandResultToJob({ + jobId: backupJob.id, + orgId: backupJob.orgId, + deviceId: backupJob.deviceId, + resultStatus: 'failed', + result: { + error: `Malformed backup result payload: ${parsedBackup.error.issues.map((issue) => issue.message).join(', ')}`, + }, + }); + } else { + await applyBackupCommandResultToJob({ + jobId: backupJob.id, + orgId: backupJob.orgId, + deviceId: backupJob.deviceId, + resultStatus: result.status, + // Provider-backed backups do not report `partial` today, but this + // path parses the agent's status and must not be the one place + // that silently discards it. + agentStatus: parsedBackup.data.status, + result: { + ...parsedBackup.data, + error: result.error || result.stderr, + }, + }); + } + } + } + } catch (err) { + console.error(`[AgentWs] Failed to process ${command.type} backup result for ${agentId}:`, err); + captureException(err); + } +} + +async function handleVaultSyncResult({ agentId, command, result, resolvedDeviceId, stdout }: Parameters[0]): Promise { + try { + await applyVaultSyncCommandResult({ + deviceId: resolvedDeviceId, + command, + resultStatus: result.status, + stdout, + stderr: result.stderr, + error: result.error, + }); + } catch (err) { + console.error(`[AgentWs] Failed to process vault sync result for ${agentId}:`, err); + captureException(err); + } +} + +async function handleSnmpPollResult({ agentId, command, result, commandId }: Parameters[0]): Promise { + try { + const payload = command.payload as Record | null; + const expectedDeviceId = typeof payload?.deviceId === 'string' ? payload.deviceId : null; + const snmpData = result.result as { + deviceId?: string; + metrics?: SnmpMetricResult[]; + } | undefined; + + if (snmpData?.deviceId && snmpData.metrics && snmpData.metrics.length > 0) { + if (!expectedDeviceId || snmpData.deviceId !== expectedDeviceId) { + console.warn( + `[AgentWs] Rejecting mismatched SNMP result ${commandId} from agent ${agentId}: ` + + `sentDevice=${snmpData.deviceId} expected=${expectedDeviceId ?? 'none'}` + ); + return; + } + if (isRedisAvailable()) { + 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}`); + const { snmpDevices } = await import('../db/schema'); + await db + .update(snmpDevices) + .set({ + lastPolled: new Date(), + // The device answered; only our own pipeline failed. Clear the + // failure backoff (#3217) so a Redis outage doesn't march every + // healthy SNMP target to 'offline' and a one-hour interval. + lastPollAttemptedAt: new Date(), + consecutiveFailures: 0, + lastStatus: 'warning' + }) + .where(eq(snmpDevices.id, expectedDeviceId)); + } + } + } catch (err) { + console.error(`[AgentWs] Failed to process SNMP poll results for ${agentId}:`, err); + captureException(err); + } +} + +async function handleScriptResult({ agentId, command, result, resolvedDeviceId, stdout }: Parameters[0]): Promise { + try { + const payload = command.payload as Record | null; + const executionId = payload?.executionId as string | undefined; + // #3162: `script_executions.id` is a uuid column, so a non-uuid + // executionId makes the UPDATE below throw with `invalid input syntax for + // type uuid` — swallowed by the catch at the bottom of this function, + // taking the agent's stdout with it. + // + // Nothing should mint a non-uuid executionId any more (the automation + // `execute_command` action, the only producer, now omits the field + // entirely). This guard is for commands queued BEFORE that deploy and still + // in flight, so it reports rather than silently skipping: a fresh non-uuid + // id means an unknown producer is sending garbage. + if (executionId && !PG_UUID_REGEX.test(executionId)) { + console.warn( + `[AgentWs] Skipping script_executions update for non-uuid executionId ${executionId} (command ${command.id})` + ); + captureException( + new Error('Non-uuid executionId in script command payload'), + undefined, + { commandId: command.id, agentId, executionId }, + ); + return; + } + if (executionId) { + let scriptStatus: 'completed' | 'failed' | 'timeout'; + if (result.status === 'completed') { + scriptStatus = result.exitCode && result.exitCode !== 0 ? 'failed' : 'completed'; + } else if (result.status === 'timeout') { + scriptStatus = 'timeout'; + } else { + scriptStatus = 'failed'; + } + + const updatedExecutions = await db + .update(scriptExecutions) + .set({ + status: scriptStatus, + completedAt: new Date(), + exitCode: result.exitCode ?? null, + // #2434: script output/errors surface to scripts:read users in the + // web UI — redact secrets before persistence (idempotent when the + // ingest chokepoint already redacted error/stderr). + stdout: stdout != null ? redactSecretsFromOutput(stdout) : null, + stderr: redactOptionalSecretText(result.stderr) ?? null, + errorMessage: redactOptionalSecretText(result.error) ?? null, + }) + .where(and( + eq(scriptExecutions.id, executionId), + eq(scriptExecutions.deviceId, resolvedDeviceId), + inArray(scriptExecutions.status, ['pending', 'queued', 'running']) + )) + .returning({ + id: scriptExecutions.id, + scriptId: scriptExecutions.scriptId, + }); + + // Update batch counters if this is part of a batch + const batchId = payload?.batchId as string | undefined; + if (batchId && updatedExecutions[0]) { + const counterField = scriptStatus === 'completed' ? 'devicesCompleted' : 'devicesFailed'; + await db + .update(scriptExecutionBatches) + .set({ + [counterField]: sql`${scriptExecutionBatches[counterField]} + 1` + }) + .where(and( + eq(scriptExecutionBatches.id, batchId), + eq(scriptExecutionBatches.scriptId, updatedExecutions[0].scriptId) + )); + } + } + } catch (err) { + // #3162 lived undetected because this catch logged to the container and + // nothing else — a swallowed 22P02 silently discarded every automation's + // script output. Report it like the SNMP handler does so the next failure + // in here (schema drift, a redaction throw, a batch-counter FK violation) + // surfaces instead of quietly eating results. + console.error(`[AgentWs] Failed to process script result for ${agentId}:`, err); + captureException(err, undefined, { + commandId: command.id, + agentId, + executionId: String((command.payload as Record | null)?.executionId ?? ''), + }); + } +} + +async function handleSensitiveDataResult({ agentId, command, result, stdout }: Parameters[0]): Promise { + try { + const { handleSensitiveDataCommandResult } = await import('../routes/agents/helpers'); + await handleSensitiveDataCommandResult(command, { + status: result.status, + exitCode: result.exitCode, + stdout, + stderr: result.stderr, + durationMs: result.durationMs, + error: result.error, + } as any); + } catch (err) { + console.error(`[AgentWs] Failed to process sensitive data result for ${agentId}:`, err); + } +} + +async function handleCisResult({ agentId, command, result, stdout }: Parameters[0]): Promise { + try { + const { handleCisCommandResult } = await import('../routes/agents/helpers'); + await handleCisCommandResult(command, { + status: result.status, + exitCode: result.exitCode, + stdout, + stderr: result.stderr, + durationMs: result.durationMs, + error: result.error, + } as any); + } catch (err) { + console.error(`[AgentWs] Failed to process CIS result for ${agentId}:`, err); + } +} + +export const commandResultHandlers: Record = { + network_discovery: handleDiscoveryResult, + backup_verify: handleBackupVerificationResult, + backup_test_restore: handleBackupVerificationResult, + backup_restore: handleVmRestoreResult, + vm_restore_from_backup: handleVmRestoreResult, + vm_instant_boot: handleVmRestoreResult, + bmr_recover: handleVmRestoreResult, + hyperv_backup: handleProviderBackedBackupResult, + mssql_backup: handleProviderBackedBackupResult, + vault_sync: handleVaultSyncResult, + snmp_poll: handleSnmpPollResult, + script: handleScriptResult, + sensitive_data_scan: handleSensitiveDataResult, + encrypt_file: handleSensitiveDataResult, + secure_delete_file: handleSensitiveDataResult, + quarantine_file: handleSensitiveDataResult, + cis_benchmark: handleCisResult, + apply_cis_remediation: handleCisResult, +};