fix(security-sync): serialize owner syncs and guard run progress

services/security-auto-analysis/src/dispatcher.test.ts

@@ -96,11 +96,93 @@ describe('dispatchDueOwners telemetry', () => {
       requeued_pending_count: 1,
       failed_running_count: 2,
     });
-    const serializedLogs = JSON.stringify(loggerMock.withTags.mock.calls);
-    expect(serializedLogs).not.toContain('sensitive-owner-id');
-    expect(serializedLogs).not.toContain('sensitive-command-id');
-    expect(serializedLogs).not.toContain('sensitive-attempt-id');
-    expect(serializedLogs).not.toContain('postgres://');
+    const serializedStageLogs = JSON.stringify(
+      loggerMock.withTags.mock.calls.filter(
+        ([tags]) =>
+          (tags as { event_name?: string }).event_name ===
+          'security_auto_analysis.dispatcher_stage_succeeded'
+      )
+    );
+    expect(serializedStageLogs).not.toContain('sensitive-owner-id');
+    expect(serializedStageLogs).not.toContain('sensitive-command-id');
+    expect(serializedStageLogs).not.toContain('sensitive-attempt-id');
+    expect(serializedStageLogs).not.toContain('postgres://');
+  });
+
+  it('logs one command_stalled event per reconciled stale command', async () => {
+    vi.mocked(reconcileStaleSecurityAgentCommands).mockResolvedValueOnce({
+      staleAccepted: [
+        {
+          id: 'stale-accepted-id',
+          command_type: 'sync',
+          owned_by_organization_id: 'org-123',
+          accepted_at: new Date(Date.now() - 5_000).toISOString(),
+          result_code: 'COMMAND_STALLED',
+          repo_full_name: 'acme/widgets',
+        },
+      ],
+      staleRunning: [
+        {
+          id: 'stale-running-id',
+          command_type: 'sync',
+          owned_by_user_id: 'user-123',
+          accepted_at: new Date(Date.now() - 9_000).toISOString(),
+          started_at: new Date(Date.now() - 4_000).toISOString(),
+          last_error_redacted: 'must-not-be-logged',
+        },
+      ],
+    } as never);
+
+    await dispatchDueOwners(env(), 'dispatch-123');
+
+    const stallTags = loggerMock.withTags.mock.calls
+      .map(([tags]) => tags as Record<string, unknown>)
+      .filter(tags => tags.event_name === 'security_auto_analysis.command_stalled');
+    expect(stallTags).toHaveLength(2);
+    expect(stallTags[0]).toMatchObject({
+      command_id: 'stale-accepted-id',
+      command_type: 'sync',
+      owner_type: 'org',
+      owner_id: 'org-123',
+      previous_status: 'accepted',
+      result_code: 'COMMAND_STALLED',
+      repo_full_name: 'acme/widgets',
+    });
+    expect(Number(stallTags[0]?.age_ms)).toBeGreaterThanOrEqual(4_000);
+    expect(stallTags[1]).toMatchObject({
+      command_id: 'stale-running-id',
+      owner_type: 'user',
+      owner_id: 'user-123',
+      previous_status: 'running',
+      result_code: null,
+    });
+    expect(Number(stallTags[1]?.age_ms)).toBeGreaterThanOrEqual(3_000);
+    const serializedStallLogs = JSON.stringify(stallTags);
+    expect(serializedStallLogs).not.toContain('must-not-be-logged');
+    expect(serializedStallLogs).not.toContain('postgres://');
+    expect(stallTags[1]?.age_ms).not.toBeNull();
+  });
+
+  it('logs command_stalled with a null age_ms when timestamps are absent', async () => {
+    vi.mocked(reconcileStaleSecurityAgentCommands).mockResolvedValueOnce({
+      staleAccepted: [{ id: 'no-timestamp-id', command_type: 'sync' }],
+      staleRunning: [],
+    } as never);
+
+    await dispatchDueOwners(env(), 'dispatch-123');
+
+    const stallTags = loggerMock.withTags.mock.calls
+      .map(([tags]) => tags as Record<string, unknown>)
+      .filter(tags => tags.event_name === 'security_auto_analysis.command_stalled');
+    expect(stallTags).toHaveLength(1);
+    expect(stallTags[0]).toMatchObject({
+      command_id: 'no-timestamp-id',
+      age_ms: null,
+      started_at: null,
+      accepted_at: null,
+      result_code: null,
+      repo_full_name: null,
+    });
   });
 
   it.each([

services/security-auto-analysis/src/dispatcher.ts

@@ -2,6 +2,7 @@ import { randomUUID } from 'crypto';
 import {
   deleteRetainedSecurityAgentCommands,
   reconcileStaleSecurityAgentCommands,
+  type SecurityAgentCommand,
 } from '@kilocode/db';
 import { getWorkerDb } from '@kilocode/db/client';
 import { discoverDueOwners, reconcileStaleAnalysisQueueRows } from './db/queries.js';
@@ -14,6 +15,40 @@ const DISPATCH_REMEDIATION_ATTEMPT_LIMIT = 100;
 const QUEUE_SEND_BATCH_LIMIT = 100;
 const DISPATCH_STAGE_SUCCEEDED_EVENT = 'security_auto_analysis.dispatcher_stage_succeeded';
 const DISPATCH_FAILED_EVENT = 'security_auto_analysis.dispatcher_failed';
+const COMMAND_STALLED_EVENT = 'security_auto_analysis.command_stalled';
+
+function logStalledCommands(reconciliation: {
+  staleAccepted: SecurityAgentCommand[];
+  staleRunning: SecurityAgentCommand[];
+}): void {
+  for (const [previousStatus, commands] of [
+    ['accepted', reconciliation.staleAccepted] as const,
+    ['running', reconciliation.staleRunning] as const,
+  ]) {
+    for (const command of commands) {
+      const timestamp =
+        previousStatus === 'running'
+          ? (command.started_at ?? command.accepted_at)
+          : command.accepted_at;
+      const parsed = timestamp ? Date.parse(timestamp) : Number.NaN;
+      logger
+        .withTags({
+          event_name: COMMAND_STALLED_EVENT,
+          command_id: command.id,
+          command_type: command.command_type,
+          owner_type: command.owned_by_organization_id ? 'org' : 'user',
+          owner_id: command.owned_by_organization_id ?? command.owned_by_user_id ?? undefined,
+          previous_status: previousStatus,
+          result_code: command.result_code ?? null,
+          started_at: command.started_at ?? null,
+          accepted_at: command.accepted_at ?? null,
+          age_ms: Number.isFinite(parsed) ? Date.now() - parsed : null,
+          repo_full_name: command.repo_full_name ?? null,
+        })
+        .info(COMMAND_STALLED_EVENT);
+    }
+  }
+}
 
 async function runDispatcherStage<T>(options: {
   stage: DispatcherStage;
@@ -100,6 +135,8 @@ export async function dispatchDueOwners(
     }),
   });
 
+  logStalledCommands(commandReconciliation);
+
   const deletedCommandCount = await runDispatcherStage({
     stage: 'retained_command_deletion',
     dispatchId,

services/security-auto-analysis/src/logger.ts

@@ -31,6 +31,16 @@ export type SecurityAutoAnalysisLogTags = {
   enqueued_owner_message_count?: number;
   discovered_remediation_attempt_count?: number;
   enqueued_remediation_message_count?: number;
+  command_id?: string;
+  command_type?: string;
+  owner_type?: 'org' | 'user';
+  owner_id?: string;
+  previous_status?: string;
+  result_code?: string | null;
+  started_at?: string | null;
+  accepted_at?: string | null;
+  age_ms?: number | null;
+  repo_full_name?: string | null;
 };
 
 export function sanitizedExceptionName(error: unknown): string {

services/security-sync/package.json

@@ -9,7 +9,8 @@
     "typecheck": "tsgo --noEmit",
     "lint": "pnpm -w exec oxlint --config .oxlintrc.json services/security-sync/src",
     "types": "wrangler types --env-interface CloudflareEnv worker-configuration.d.ts",
-    "test": "vitest run"
+    "test": "vitest run",
+    "test:integration": "vitest run --config vitest.integration.config.ts"
   },
   "dependencies": {
     "@kilocode/db": "workspace:*",