feat(security-agent): move manual workflows into workers (#3312)

Co-authored-by: kiloconnect[bot] <240665456+kiloconnect[bot]@users.noreply.github.com>

Author: jeanduplessis

apps/web/.env.development.local.example

@@ -16,6 +16,12 @@ AUTO_FIX_URL=http://localhost:8792
 # @url cloudflare-auto-triage-infra
 AUTO_TRIAGE_URL=http://localhost:8791
 
+# @url cloudflare-security-sync
+SECURITY_SYNC_WORKER_URL=http://localhost:8812
+
+# @url cloudflare-security-auto-analysis
+SECURITY_AUTO_ANALYSIS_WORKER_URL=http://localhost:8797
+
 # @url cloudflare-app-builder
 APP_BUILDER_URL=http://localhost:8790
 

apps/web/src/app/api/internal/security-analysis-callback/[findingId]/route.test.ts

@@ -105,10 +105,18 @@ jest.mock('@/lib/utils.server', () => ({
 }));
 
 jest.mock('@/lib/drizzle', () => {
+  let selectedTable: unknown;
   const chain = {
-    from: jest.fn().mockReturnThis(),
+    from: jest.fn((table: unknown) => {
+      selectedTable = table;
+      return chain;
+    }),
     where: jest.fn().mockReturnThis(),
-    limit: jest.fn(() => mockDbSelect()),
+    limit: jest.fn(() =>
+      (selectedTable as { __name?: string } | undefined)?.__name === 'security_analysis_queue'
+        ? Promise.resolve([{ claimToken: 'attempt-token-123' }])
+        : mockDbSelect()
+    ),
   };
   return {
     db: {
@@ -119,16 +127,25 @@ jest.mock('@/lib/drizzle', () => {
 
 jest.mock('@kilocode/db/schema', () => ({
   kilocode_users: { id: 'id' },
+  security_analysis_queue: {
+    __name: 'security_analysis_queue',
+    claim_token: 'claim_token',
+    finding_id: 'finding_id',
+    queue_status: 'queue_status',
+  },
 }));
 
 jest.mock('drizzle-orm', () => ({
+  and: jest.fn(),
   eq: jest.fn(),
+  inArray: jest.fn(),
 }));
 
 // --- Helpers ---
 
 const CALLBACK_SECRET = 'test-callback-token-secret';
 const FINDING_ID = 'finding-abc-123';
+const ATTEMPT_TOKEN = 'attempt-token-123';
 let defaultCallbackToken: string;
 
 function makeRequest(
@@ -137,6 +154,9 @@ function makeRequest(
   callbackToken: string | null = defaultCallbackToken
 ): NextRequest {
   return {
+    nextUrl: new URL(
+      `https://app.kilo.ai/api/internal/security-analysis-callback/${findingId}?attempt=${ATTEMPT_TOKEN}`
+    ),
     headers: {
       get: (name: string) => {
         if (name === 'X-Callback-Token') return callbackToken;
@@ -248,7 +268,7 @@ beforeEach(async () => {
   defaultCallbackToken = await deriveCallbackToken({
     secret: CALLBACK_SECRET,
     scope: 'security-analysis-callback',
-    resourceParts: [FINDING_ID],
+    resourceParts: [FINDING_ID, ATTEMPT_TOKEN],
   });
   mockUpdateAnalysisStatus.mockResolvedValue(true);
   mockTransitionAutoAnalysisQueueFromCallback.mockResolvedValue(undefined);
@@ -292,7 +312,7 @@ describe('POST /api/internal/security-analysis-callback/[findingId]', () => {
       const callbackToken = await deriveCallbackToken({
         secret: CALLBACK_SECRET,
         scope: 'security-analysis-callback',
-        resourceParts: ['different-finding'],
+        resourceParts: ['different-finding', ATTEMPT_TOKEN],
       });
       const req = makeRequest(FINDING_ID, completedPayload, callbackToken);
       const response = await POST(req, makeParams(FINDING_ID));
@@ -366,6 +386,7 @@ describe('POST /api/internal/security-analysis-callback/[findingId]', () => {
       expect(body.message).toBe('Superseded finding ignored');
       expect(mockTransitionAutoAnalysisQueueFromCallback).toHaveBeenCalledWith({
         findingId: FINDING_ID,
+        attemptToken: ATTEMPT_TOKEN,
         toStatus: 'completed',
         failureCode: 'SKIPPED_NO_LONGER_ELIGIBLE',
       });

apps/web/src/app/api/internal/security-analysis-callback/[findingId]/route.ts

@@ -17,8 +17,8 @@ import { fetchSessionSnapshot } from '@/lib/session-ingest-client';
 import { trackSecurityAgentAnalysisCompleted } from '@/lib/security-agent/posthog-tracking';
 import { generateApiToken } from '@/lib/tokens';
 import { db } from '@/lib/drizzle';
-import { kilocode_users } from '@kilocode/db/schema';
-import { eq } from 'drizzle-orm';
+import { kilocode_users, security_analysis_queue } from '@kilocode/db/schema';
+import { and, eq, inArray } from 'drizzle-orm';
 import { z } from 'zod';
 import { verifyCallbackToken } from '@kilocode/worker-utils/callback-token';
 import { logExceptInTest, sentryLogger } from '@/lib/utils.server';
@@ -32,6 +32,8 @@ import {
   DEFAULT_SECURITY_AGENT_TRIAGE_MODEL,
 } from '@/lib/security-agent/core/constants';
 
+// Compatibility-only callback ingress retained for explicit rollback routing.
+// Durable default ingress lives in the security-auto-analysis Worker.
 const log = sentryLogger('security-agent:callback', 'info');
 const warn = sentryLogger('security-agent:callback', 'warning');
 const logError = sentryLogger('security-agent:callback', 'error');
@@ -82,14 +84,18 @@ export async function POST(
 ) {
   try {
     const { findingId } = await params;
+    const attemptToken = req.nextUrl.searchParams.get('attempt');
+    if (!attemptToken) {
+      return NextResponse.json({ error: 'Missing callback attempt token' }, { status: 400 });
+    }
     const callbackToken = req.headers.get('X-Callback-Token');
     const validCallbackToken =
       !!CALLBACK_TOKEN_SECRET &&
       (await verifyCallbackToken({
         token: callbackToken,
         secret: CALLBACK_TOKEN_SECRET,
         scope: 'security-analysis-callback',
-        resourceParts: [findingId],
+        resourceParts: [findingId, attemptToken],
       }));
     if (!validCallbackToken) {
       return NextResponse.json({ error: 'Unauthorized' }, { status: 401 });
@@ -117,6 +123,28 @@ export async function POST(
       return NextResponse.json({ error: 'Finding not found' }, { status: 404 });
     }
 
+    const [activeAttempt] = await db
+      .select({ claimToken: security_analysis_queue.claim_token })
+      .from(security_analysis_queue)
+      .where(
+        and(
+          eq(security_analysis_queue.finding_id, findingId),
+          inArray(security_analysis_queue.queue_status, ['pending', 'running'])
+        )
+      )
+      .limit(1);
+    if (
+      (finding.analysis_status === 'pending' || finding.analysis_status === 'running') &&
+      activeAttempt?.claimToken !== attemptToken
+    ) {
+      warn('Ignoring stale auto-analysis callback due to attempt mismatch', {
+        findingId,
+        callbackAttemptToken: attemptToken,
+        activeAttemptToken: activeAttempt?.claimToken ?? null,
+      });
+      return NextResponse.json({ success: true, message: 'Stale callback ignored' });
+    }
+
     const sessionMismatch =
       (payload.cloudAgentSessionId &&
         finding.session_id &&
@@ -158,6 +186,7 @@ export async function POST(
       });
       await transitionAutoAnalysisQueueFromCallback({
         findingId,
+        attemptToken,
         toStatus: 'completed',
         failureCode: 'SKIPPED_NO_LONGER_ELIGIBLE',
       });
@@ -184,9 +213,9 @@ export async function POST(
     after(async () => {
       try {
         if (payload.status === 'completed') {
-          await handleAnalysisCompleted(findingId, payload, finding);
+          await handleAnalysisCompleted(findingId, attemptToken, payload, finding);
         } else if (payload.status === 'failed' || payload.status === 'interrupted') {
-          await handleAnalysisFailed(findingId, payload, finding);
+          await handleAnalysisFailed(findingId, attemptToken, payload, finding);
         } else {
           const unknownStatus = payload.status as string;
           logError('Unknown callback status received, marking as failed', {
@@ -202,6 +231,7 @@ export async function POST(
           }
           await transitionAutoAnalysisQueueFromCallback({
             findingId,
+            attemptToken,
             toStatus: 'failed',
             failureCode: 'STATE_GUARD_REJECTED',
             errorMessage: `Unknown callback status: ${unknownStatus}`,
@@ -255,6 +285,7 @@ function readAnalysisContext(analysis: SecurityFindingAnalysis | null | undefine
 
 async function handleAnalysisCompleted(
   findingId: string,
+  attemptToken: string,
   payload: ExecutionCallbackPayload,
   finding: Awaited<ReturnType<typeof getSecurityFindingById>> & {}
 ) {
@@ -281,6 +312,7 @@ async function handleAnalysisCompleted(
     }
     await transitionAutoAnalysisQueueFromCallback({
       findingId,
+      attemptToken,
       toStatus: 'failed',
       failureCode: 'STATE_GUARD_REJECTED',
       errorMessage: 'Cannot process callback — triggeredByUserId missing from analysis context',
@@ -300,6 +332,7 @@ async function handleAnalysisCompleted(
     }
     await transitionAutoAnalysisQueueFromCallback({
       findingId,
+      attemptToken,
       toStatus: 'failed',
       failureCode: 'STATE_GUARD_REJECTED',
       errorMessage: 'Callback missing kiloSessionId — cannot retrieve analysis result',
@@ -365,6 +398,7 @@ async function handleAnalysisCompleted(
     }
     await transitionAutoAnalysisQueueFromCallback({
       findingId,
+      attemptToken,
       toStatus: 'failed',
       failureCode: 'START_CALL_AMBIGUOUS',
       errorMessage: 'Analysis completed but result could not be retrieved from ingest service',
@@ -404,6 +438,7 @@ async function handleAnalysisCompleted(
     }
     await transitionAutoAnalysisQueueFromCallback({
       findingId,
+      attemptToken,
       toStatus: 'failed',
       failureCode: 'STATE_GUARD_REJECTED',
       errorMessage: `User ${triggeredByUserId} not found — cannot run Tier 3 extraction`,
@@ -449,6 +484,7 @@ async function handleAnalysisCompleted(
     });
     await transitionAutoAnalysisQueueFromCallback({
       findingId,
+      attemptToken,
       toStatus: 'failed',
       failureCode: 'START_CALL_AMBIGUOUS',
       errorMessage: error instanceof Error ? error.message : String(error),
@@ -458,17 +494,23 @@ async function handleAnalysisCompleted(
 
   const updatedFinding = await getSecurityFindingById(findingId);
   if (updatedFinding?.analysis_status === 'completed') {
-    await transitionAutoAnalysisQueueFromCallback({ findingId, toStatus: 'completed' });
+    await transitionAutoAnalysisQueueFromCallback({
+      findingId,
+      attemptToken,
+      toStatus: 'completed',
+    });
   } else if (updatedFinding?.analysis_status === 'failed') {
     await transitionAutoAnalysisQueueFromCallback({
       findingId,
+      attemptToken,
       toStatus: 'failed',
       failureCode: 'START_CALL_AMBIGUOUS',
       errorMessage: updatedFinding.analysis_error ?? undefined,
     });
   } else {
     await transitionAutoAnalysisQueueFromCallback({
       findingId,
+      attemptToken,
       toStatus: 'failed',
       failureCode: 'STATE_GUARD_REJECTED',
       errorMessage: `Unexpected post-finalize state: ${updatedFinding?.analysis_status ?? 'finding_not_found'}`,
@@ -478,6 +520,7 @@ async function handleAnalysisCompleted(
 
 async function handleAnalysisFailed(
   findingId: string,
+  attemptToken: string,
   payload: ExecutionCallbackPayload,
   finding: Awaited<ReturnType<typeof getSecurityFindingById>> & {}
 ) {
@@ -521,6 +564,7 @@ async function handleAnalysisFailed(
   }
   await transitionAutoAnalysisQueueFromCallback({
     findingId,
+    attemptToken,
     toStatus: 'failed',
     failureCode: callbackFailure.failureCode,
     errorMessage,

apps/web/src/components/security-agent/FindingDetailDialog.tsx

@@ -29,6 +29,8 @@ import type { SecurityFinding } from '@kilocode/db/schema';
 import { useTRPC } from '@/lib/trpc/utils';
 import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query';
 import Link from 'next/link';
+import { toast } from 'sonner';
+import { manualAnalysisAdmissionCopy } from './manual-analysis-admission-copy';
 
 type Severity = 'critical' | 'high' | 'medium' | 'low';
 
@@ -106,6 +108,7 @@ export function FindingDetailDialog({
   const startOrgAnalysisMutation = useMutation(
     trpc.organizations.securityAgent.startAnalysis.mutationOptions({
       onSuccess: async () => {
+        toast.success(manualAnalysisAdmissionCopy.successTitle);
         await queryClient.invalidateQueries();
       },
     })
@@ -115,6 +118,7 @@ export function FindingDetailDialog({
   const startUserAnalysisMutation = useMutation(
     trpc.securityAgent.startAnalysis.mutationOptions({
       onSuccess: async () => {
+        toast.success(manualAnalysisAdmissionCopy.successTitle);
         await queryClient.invalidateQueries();
       },
     })