diff --git a/apps/web/src/app/admin/deletion-queue/[id]/DeletionQueueDetailContent.tsx b/apps/web/src/app/admin/deletion-queue/[id]/DeletionQueueDetailContent.tsx index af82352aae..ca0d1441ee 100644 --- a/apps/web/src/app/admin/deletion-queue/[id]/DeletionQueueDetailContent.tsx +++ b/apps/web/src/app/admin/deletion-queue/[id]/DeletionQueueDetailContent.tsx @@ -39,6 +39,8 @@ import { deletionAttentionHint } from '@/lib/user/deletion-queue/deletion-hints' import { deletionStepDescription, deletionStepLabel, + deletionStepProgressLabel, + formatActivityDetail, formatAge, formatTimestamp, humanizeToken, @@ -304,9 +306,7 @@ export function DeletionQueueDetailContent({

- {[item.stepKey, item.details.errorCode, item.details.httpStatusClass] - .filter(Boolean) - .join(' · ') || '—'} + {formatActivityDetail(item)}

)) @@ -369,7 +369,6 @@ function CompactDeletionDetail({ }) { const request = detail.request; const ticket = request.pylonTicket ? `#${request.pylonTicket.replace(/^#/, '')}` : null; - const currentTask = detail.tasks.find(task => !isFinishedTask(task.status)); const stuckTask = detail.tasks.find( task => task.status === 'needs_attention' || task.status === 'manual_action_required' ); @@ -457,6 +456,7 @@ function CompactDeletionDetail({ return task ? [task] : []; }); if (tasks.length === 0) return null; + const unlocked = isProgressGroupUnlocked(detail.tasks, groupIndex); return (
{groupIndex > 0 ? ( @@ -479,7 +479,7 @@ function CompactDeletionDetail({ ))}
@@ -506,11 +506,7 @@ function CompactDeletionDetail({ {humanizeToken(item.eventType)} {formatTimestamp(item.createdAt)} -

- {[item.stepKey ? deletionStepLabel(item.stepKey) : null, item.details.errorCode] - .filter(Boolean) - .join(' · ') || '—'} -

+

{formatActivityDetail(item)}

)) )} @@ -598,9 +594,39 @@ function isFinishedTask(status: string): boolean { return status === 'succeeded' || status === 'not_applicable' || status === 'manually_verified'; } +function isStuckTask(status: string): boolean { + return status === 'needs_attention' || status === 'manual_action_required'; +} + +function isOpenTask(status: string): boolean { + return !isFinishedTask(status) && !isStuckTask(status); +} + +function isProgressGroupUnlocked(tasks: Task[], groupIndex: number): boolean { + return PROGRESS_GROUPS.slice(0, groupIndex).every(group => + group.stepKeys.every(stepKey => { + const task = tasks.find(item => item.stepKey === stepKey); + return !task || isFinishedTask(task.status); + }) + ); +} + function ProgressStepTile({ task, current }: { task: Task; current: boolean }) { const finished = isFinishedTask(task.status); - const stuck = task.status === 'needs_attention' || task.status === 'manual_action_required'; + const stuck = isStuckTask(task.status); + const countLabel = deletionStepProgressLabel( + task.stepKey, + task.processedCount, + task.scannedCount + ); + const description = + stuck && task.lastErrorCode + ? task.lastErrorCode + : current && countLabel + ? `${countLabel} so far` + : finished && countLabel + ? countLabel + : deletionStepDescription(task.stepKey); return (
- {finished ? : current && !finished ? '▸' : '·'} + {finished ? : current ? '▸' : '·'}

{deletionStepLabel(task.stepKey)}

-

- {stuck && task.lastErrorCode ? task.lastErrorCode : deletionStepDescription(task.stepKey)} -

+

{description}

); diff --git a/apps/web/src/app/admin/deletion-queue/deletion-queue-format.test.ts b/apps/web/src/app/admin/deletion-queue/deletion-queue-format.test.ts index cd30e17e70..9e80e3f289 100644 --- a/apps/web/src/app/admin/deletion-queue/deletion-queue-format.test.ts +++ b/apps/web/src/app/admin/deletion-queue/deletion-queue-format.test.ts @@ -1,4 +1,10 @@ -import { parseDeletionEntries, parseDeletionQueueTab } from './deletion-queue-format'; +import { + deletionStepCountLabel, + deletionStepProgressLabel, + formatActivityDetail, + parseDeletionEntries, + parseDeletionQueueTab, +} from './deletion-queue-format'; describe('parseDeletionEntries', () => { it('parses one email per line', () => { @@ -50,6 +56,63 @@ describe('parseDeletionEntries', () => { }); }); +describe('deletionStepCountLabel', () => { + it('uses a step-specific verb for known cleanup tasks', () => { + expect(deletionStepCountLabel('cli_v2_sessions', 12)).toBe('12 deleted'); + expect(deletionStepCountLabel('usage_prompt_prefixes', 340)).toBe('340 scrubbed'); + expect(deletionStepCountLabel('kiloclaw_destroy', 2)).toBe('2 destroyed'); + expect(deletionStepCountLabel('customerio', 1)).toBe('1 removed'); + }); +}); + +describe('deletionStepProgressLabel', () => { + it('shows scanned usage rows even when nothing was scrubbed', () => { + expect(deletionStepProgressLabel('usage_prompt_prefixes', 80, 49000)).toBe( + '80 scrubbed · 49000 scanned' + ); + expect(deletionStepProgressLabel('usage_prompt_prefixes', 0, 1000)).toBe('1000 scanned'); + expect(deletionStepProgressLabel('cli_v2_sessions', 0, 0)).toBeNull(); + }); +}); + +describe('formatActivityDetail', () => { + it('shows the step and how many records were processed', () => { + expect( + formatActivityDetail({ + stepKey: 'cli_v1_blobs', + details: { processedCount: 3, errorCode: null }, + }) + ).toBe('CLI v1 · 3 deleted'); + }); + + it('includes zero counts so empty work is visible', () => { + expect( + formatActivityDetail({ + stepKey: 'cli_v2_sessions', + details: { processedCount: 0, errorCode: null }, + }) + ).toBe('CLI sessions · 0 deleted'); + }); + + it('keeps error codes next to the count', () => { + expect( + formatActivityDetail({ + stepKey: 'usage_prompt_prefixes', + details: { processedCount: 40, errorCode: 'usage_prefix_page_timeout' }, + }) + ).toBe('Usage prompts · 40 scrubbed · usage_prefix_page_timeout'); + }); + + it('shows scanned usage rows next to scrubbed prefixes', () => { + expect( + formatActivityDetail({ + stepKey: 'usage_prompt_prefixes', + details: { processedCount: 80, scannedCount: 49000, errorCode: null }, + }) + ).toBe('Usage prompts · 80 scrubbed · 49000 scanned'); + }); +}); + describe('parseDeletionQueueTab', () => { it('accepts the remaining tabs and falls back unknown values to open', () => { expect(parseDeletionQueueTab('open')).toBe('open'); diff --git a/apps/web/src/app/admin/deletion-queue/deletion-queue-format.ts b/apps/web/src/app/admin/deletion-queue/deletion-queue-format.ts index 5cc8236515..78e04b858a 100644 --- a/apps/web/src/app/admin/deletion-queue/deletion-queue-format.ts +++ b/apps/web/src/app/admin/deletion-queue/deletion-queue-format.ts @@ -72,6 +72,58 @@ export function deletionStepDescription(stepKey: string): string { return STEP_LABELS[stepKey]?.description ?? ''; } +export function deletionStepCountLabel(stepKey: string, count: number): string { + switch (stepKey) { + case 'usage_prompt_prefixes': + return `${count} scrubbed`; + case 'kiloclaw_destroy': + return `${count} destroyed`; + case 'customerio': + case 'substack': + return `${count} removed`; + default: + return `${count} deleted`; + } +} + +export function deletionStepProgressLabel( + stepKey: string, + processedCount: number, + scannedCount = 0 +): string | null { + const parts: string[] = []; + if (processedCount > 0) parts.push(deletionStepCountLabel(stepKey, processedCount)); + if (scannedCount > 0) parts.push(`${scannedCount} scanned`); + return parts.length > 0 ? parts.join(' · ') : null; +} + +export function formatActivityDetail(item: { + stepKey: string | null; + details: { + processedCount: number | null; + scannedCount?: number | null; + errorCode: string | null; + httpStatusClass?: string | null; + }; +}): string { + const count = + item.stepKey && item.details.processedCount != null + ? deletionStepCountLabel(item.stepKey, item.details.processedCount) + : null; + const scanned = item.details.scannedCount != null ? `${item.details.scannedCount} scanned` : null; + return ( + [ + item.stepKey ? deletionStepLabel(item.stepKey) : null, + count, + scanned, + item.details.errorCode, + item.details.httpStatusClass, + ] + .filter(Boolean) + .join(' · ') || '—' + ); +} + export function shortId(id: string): string { return id.length > 8 ? `${id.slice(0, 8)}…` : id; } diff --git a/apps/web/src/lib/user/deletion-queue/deletion-outcomes.test.ts b/apps/web/src/lib/user/deletion-queue/deletion-outcomes.test.ts index 1f8da0d9d4..56c889d6ff 100644 --- a/apps/web/src/lib/user/deletion-queue/deletion-outcomes.test.ts +++ b/apps/web/src/lib/user/deletion-queue/deletion-outcomes.test.ts @@ -16,12 +16,60 @@ import { cleanupDbForTest, db } from '@/lib/drizzle'; import { enqueueUserDeletionTargets } from '@/lib/user/deletion-queue/deletion-enqueue'; import { markTaskManuallyVerified, + persistHandlerOutcome, persistRejectedPreflight, retryBlockedPreflight, } from '@/lib/user/deletion-queue/deletion-outcomes'; import { runDeletionPreflight } from '@/lib/user/deletion-queue/deletion-preflight'; import { insertTestUser } from '@/tests/helpers/user.helper'; +describe('persistHandlerOutcome progress', () => { + beforeEach(async () => { + await cleanupDbForTest(); + }); + + it('writes retry progress onto the step', async () => { + const { requestId, claimToken } = await enqueueRunningStep(UserDeletionStepKey.CliV2Sessions); + + const result = await persistHandlerOutcome({ + requestId, + stepKey: UserDeletionStepKey.CliV2Sessions, + claimToken, + outcome: { + kind: 'retry', + errorCode: 'http_500', + httpStatusClass: '5xx', + progress: { processed_count: 10 }, + }, + }); + + expect(result.kind).toBe('applied'); + const step = await loadStep(requestId, UserDeletionStepKey.CliV2Sessions); + expect(step?.status).toBe(UserDeletionStepStatus.RetryWait); + expect(step?.progress_json).toEqual({ processed_count: 10 }); + }); + + it('writes needs_attention progress onto the step', async () => { + const { requestId, claimToken } = await enqueueRunningStep(UserDeletionStepKey.CliV2Sessions); + + const result = await persistHandlerOutcome({ + requestId, + stepKey: UserDeletionStepKey.CliV2Sessions, + claimToken, + outcome: { + kind: 'needs_attention', + errorCode: 'session_identity_mismatch', + progress: { processed_count: 7 }, + }, + }); + + expect(result.kind).toBe('applied'); + const step = await loadStep(requestId, UserDeletionStepKey.CliV2Sessions); + expect(step?.status).toBe(UserDeletionStepStatus.NeedsAttention); + expect(step?.progress_json).toEqual({ processed_count: 7 }); + }); +}); + describe('markTaskManuallyVerified', () => { beforeEach(async () => { await cleanupDbForTest(); @@ -288,6 +336,38 @@ describe('shared preflight outcomes', () => { }); }); +async function enqueueRunningStep(stepKey: UserDeletionStepKey) { + const admin = await insertTestUser({ is_admin: true }); + const user = await insertTestUser({ + google_user_email: `running-${stepKey}-${crypto.randomUUID()}@example.com`, + }); + const [result] = await enqueueUserDeletionTargets({ + actor: { kiloUserId: admin.id }, + targets: [{ email: user.google_user_email, trustedUserId: user.id }], + }); + expect(result.status).toBe('enqueued'); + if (result.status !== 'enqueued') throw new Error('expected enqueued'); + const claimToken = crypto.randomUUID(); + await db + .update(user_deletion_requests) + .set({ status: UserDeletionRequestStatus.InProgress }) + .where(eq(user_deletion_requests.id, result.requestId)); + await db + .update(user_deletion_steps) + .set({ + status: UserDeletionStepStatus.Running, + claim_token: claimToken, + claimed_until: new Date(Date.now() + 60_000).toISOString(), + }) + .where( + and( + eq(user_deletion_steps.request_id, result.requestId), + eq(user_deletion_steps.step_key, stepKey) + ) + ); + return { requestId: result.requestId, claimToken }; +} + async function enqueueStuckStep(params: { stepKey: UserDeletionStepKey; status: UserDeletionStepStatus; diff --git a/apps/web/src/lib/user/deletion-queue/deletion-outcomes.ts b/apps/web/src/lib/user/deletion-queue/deletion-outcomes.ts index fc3798e865..4dbf4adf62 100644 --- a/apps/web/src/lib/user/deletion-queue/deletion-outcomes.ts +++ b/apps/web/src/lib/user/deletion-queue/deletion-outcomes.ts @@ -13,6 +13,7 @@ import { UserDeletionStepKey, UserDeletionStepStatus, type UserDeletionManualEvidence, + type UserDeletionTaskProgress, } from '@kilocode/db/schema-types'; import { db, type DrizzleTransaction } from '@/lib/drizzle'; import { anonymizeCloudUserData } from '@/lib/user'; @@ -291,7 +292,10 @@ async function persistTaskDispositionTx( requestId: request.id, stepKey, eventType: 'continue', - details: { processed_count: outcome.progress?.processed_count }, + details: { + processed_count: outcome.progress?.processed_count, + scanned_count: outcome.progress?.scanned_count, + }, }); break; case 'retry': { @@ -306,6 +310,7 @@ async function persistTaskDispositionTx( errorCode: outcome.errorCode, windowAttempt: nextWindow, lifetimeAttempt: nextLifetime, + progress: outcome.progress, }); break; } @@ -321,6 +326,7 @@ async function persistTaskDispositionTx( lifetime_attempt_count: nextLifetime, last_error_code: outcome.errorCode, rate_limited_since: null, + progress_json: outcome.progress ?? step.progress_json, }) .where(eq(user_deletion_steps.id, step.id)); await writeDeletionActivity(tx, { @@ -350,6 +356,7 @@ async function persistTaskDispositionTx( windowAttempt: step.window_attempt_count, lifetimeAttempt: step.lifetime_attempt_count, rateLimitedSince: step.rate_limited_since, + progress: outcome.progress, }); break; } @@ -366,6 +373,7 @@ async function persistTaskDispositionTx( claimed_until: null, last_error_code: 'rate_limited', rate_limited_since: step.rate_limited_since ?? now, + progress_json: outcome.progress ?? step.progress_json, }) .where(eq(user_deletion_steps.id, step.id)); await writeDeletionActivity(tx, { @@ -389,6 +397,7 @@ async function persistTaskDispositionTx( resourceHmac: outcome.resourceHmac, windowAttempt: step.window_attempt_count, lifetimeAttempt: step.lifetime_attempt_count, + progress: outcome.progress, }); break; case 'manual_action_required': @@ -451,6 +460,7 @@ async function persistTaskDispositionTx( details: { processed_count: outcome.kind === 'succeeded' ? outcome.progress?.processed_count : undefined, + scanned_count: outcome.kind === 'succeeded' ? outcome.progress?.scanned_count : undefined, }, }); break; @@ -765,6 +775,7 @@ async function moveToAttention( windowAttempt: number; lifetimeAttempt: number; rateLimitedSince?: string | null; + progress?: UserDeletionTaskProgress; } ): Promise { await tx @@ -777,6 +788,7 @@ async function moveToAttention( lifetime_attempt_count: params.lifetimeAttempt, last_error_code: params.errorCode, rate_limited_since: params.rateLimitedSince ?? null, + ...(params.progress ? { progress_json: params.progress } : {}), }) .where(eq(user_deletion_steps.id, params.stepId)); await writeDeletionAudit(tx, { diff --git a/apps/web/src/lib/user/deletion-queue/deletion-types.ts b/apps/web/src/lib/user/deletion-queue/deletion-types.ts index 3f8549d0fa..10bb13db99 100644 --- a/apps/web/src/lib/user/deletion-queue/deletion-types.ts +++ b/apps/web/src/lib/user/deletion-queue/deletion-types.ts @@ -10,17 +10,20 @@ export type DeletionHandlerRetry = { kind: 'retry'; errorCode: string; httpStatusClass?: string; + progress?: UserDeletionTaskProgress; }; export type DeletionHandlerRateLimited = { kind: 'rate_limited'; retryAfterMs: number; + progress?: UserDeletionTaskProgress; }; export type DeletionHandlerNeedsAttention = { kind: 'needs_attention'; errorCode: string; resourceHmac?: string; + progress?: UserDeletionTaskProgress; }; export type DeletionHandlerManualAction = { diff --git a/apps/web/src/lib/user/deletion-queue/handlers/cli-v2.test.ts b/apps/web/src/lib/user/deletion-queue/handlers/cli-v2.test.ts new file mode 100644 index 0000000000..b6f67bde33 --- /dev/null +++ b/apps/web/src/lib/user/deletion-queue/handlers/cli-v2.test.ts @@ -0,0 +1,305 @@ +jest.mock('@/lib/config.server', () => { + const actual: Record = jest.requireActual('@/lib/config.server'); + return { + ...actual, + SESSION_INGEST_WORKER_URL: 'https://test-ingest.example.com', + }; +}); + +import { and, eq } from 'drizzle-orm'; +import { + cli_sessions_v2, + type UserDeletionRequest, + type UserDeletionStep, +} from '@kilocode/db/schema'; +import { + UserDeletionCloudSubjectResolution, + UserDeletionStepKey, + UserDeletionStepStatus, +} from '@kilocode/db/schema-types'; +import { cleanupDbForTest, db } from '@/lib/drizzle'; +import { USER_DELETION_RESOURCE_BATCH_SIZE } from '@/lib/user/deletion-queue/deletion-constants'; +import type { DeletionHandlerContext } from '@/lib/user/deletion-queue/deletion-types'; +import { handleCliV2Sessions } from '@/lib/user/deletion-queue/handlers/cli-v2'; +import { insertTestUser } from '@/tests/helpers/user.helper'; + +const INGEST_BASE = 'https://test-ingest.example.com/api/session/'; + +describe('handleCliV2Sessions', () => { + beforeEach(async () => { + await cleanupDbForTest(); + }); + + afterEach(() => { + jest.restoreAllMocks(); + }); + + it('returns not_applicable when the user has no CLI v2 sessions', async () => { + const user = await insertTestUser(); + const outcome = await handleCliV2Sessions({ + request: { user_id: user.id } as UserDeletionRequest, + step: runningStep(), + context: handlerContext(), + }); + + expect(outcome).toEqual({ kind: 'not_applicable' }); + }); + + it('deletes a parent after its children across batches in one claim', async () => { + const user = await insertTestUser(); + const parentId = newSessionId('parent'); + const childA = newSessionId('childa'); + const childB = newSessionId('childb'); + await insertSession(user.id, parentId); + await insertSession(user.id, childA, parentId); + await insertSession(user.id, childB, parentId); + + const deleted = mockIngestDelete(user.id); + + const outcome = await handleCliV2Sessions({ + request: { user_id: user.id } as UserDeletionRequest, + step: runningStep(), + context: handlerContext(), + }); + + expect(outcome).toEqual({ kind: 'succeeded', progress: { processed_count: 3 } }); + expect(deleted.sort()).toEqual([childA, childB, parentId].sort()); + expect(await remainingSessionIds(user.id)).toEqual([]); + }); + + it('drains more than one leaf batch before succeeding', async () => { + const user = await insertTestUser(); + const count = USER_DELETION_RESOURCE_BATCH_SIZE + 1; + const sessionIds = Array.from({ length: count }, (_, index) => newSessionId(`b${index}`)); + for (const sessionId of sessionIds) { + await insertSession(user.id, sessionId); + } + + mockIngestDelete(user.id); + + const outcome = await handleCliV2Sessions({ + request: { user_id: user.id } as UserDeletionRequest, + step: runningStep(), + context: handlerContext(), + }); + + expect(outcome).toEqual({ kind: 'succeeded', progress: { processed_count: count } }); + expect(await remainingSessionIds(user.id)).toEqual([]); + }); + + it('deletes a leaf batch in parallel', async () => { + const user = await insertTestUser(); + const sessionIds = [newSessionId('p0'), newSessionId('p1'), newSessionId('p2')]; + for (const sessionId of sessionIds) { + await insertSession(user.id, sessionId); + } + + let inFlight = 0; + let maxInFlight = 0; + jest.spyOn(globalThis, 'fetch').mockImplementation(async input => { + const sessionId = sessionIdFromUrl(input); + inFlight += 1; + maxInFlight = Math.max(maxInFlight, inFlight); + await new Promise(resolve => setTimeout(resolve, 20)); + inFlight -= 1; + await deleteSessionRow(user.id, sessionId); + return new Response(JSON.stringify({ success: true }), { status: 200 }); + }); + + const outcome = await handleCliV2Sessions({ + request: { user_id: user.id } as UserDeletionRequest, + step: runningStep(), + context: handlerContext(), + }); + + expect(outcome).toEqual({ kind: 'succeeded', progress: { processed_count: 3 } }); + expect(maxInFlight).toBe(3); + }); + + it('continues after a 409 without dropping already-deleted siblings', async () => { + const user = await insertTestUser(); + const keepId = newSessionId('keep'); + const dropId = newSessionId('drop'); + await insertSession(user.id, keepId); + await insertSession(user.id, dropId); + + jest.spyOn(globalThis, 'fetch').mockImplementation(async input => { + const sessionId = sessionIdFromUrl(input); + if (sessionId === keepId) { + return new Response(JSON.stringify({ success: false, error: 'session_not_leaf' }), { + status: 409, + }); + } + await deleteSessionRow(user.id, sessionId); + return new Response(JSON.stringify({ success: true }), { status: 200 }); + }); + + const outcome = await handleCliV2Sessions({ + request: { user_id: user.id } as UserDeletionRequest, + step: runningStep(), + context: handlerContext(), + }); + + expect(outcome).toEqual({ kind: 'continue', progress: { processed_count: 1 } }); + expect(await remainingSessionIds(user.id)).toEqual([keepId]); + }); + + it('stops starting another batch when the cron reserve is gone', async () => { + const user = await insertTestUser(); + const count = USER_DELETION_RESOURCE_BATCH_SIZE + 1; + for (let index = 0; index < count; index += 1) { + await insertSession(user.id, newSessionId(`t${index}`)); + } + + let remainingMs = 60_000; + mockIngestDelete(user.id, () => { + remainingMs = 0; + }); + + const outcome = await handleCliV2Sessions({ + request: { user_id: user.id } as UserDeletionRequest, + step: runningStep(), + context: handlerContext(() => remainingMs), + }); + + expect(outcome).toEqual({ + kind: 'continue', + progress: { processed_count: USER_DELETION_RESOURCE_BATCH_SIZE }, + }); + expect(await remainingSessionIds(user.id)).toHaveLength(1); + }); + + it('keeps processed progress when a later batch fails', async () => { + const user = await insertTestUser(); + const count = USER_DELETION_RESOURCE_BATCH_SIZE + 1; + for (let index = 0; index < count; index += 1) { + await insertSession(user.id, newSessionId(`f${index}`)); + } + + let deleted = 0; + jest.spyOn(globalThis, 'fetch').mockImplementation(async input => { + const sessionId = sessionIdFromUrl(input); + if (deleted >= USER_DELETION_RESOURCE_BATCH_SIZE) { + return new Response('unavailable', { status: 500 }); + } + deleted += 1; + await deleteSessionRow(user.id, sessionId); + return new Response(JSON.stringify({ success: true }), { status: 200 }); + }); + + const outcome = await handleCliV2Sessions({ + request: { user_id: user.id } as UserDeletionRequest, + step: runningStep(), + context: handlerContext(), + }); + + expect(outcome).toEqual({ + kind: 'retry', + errorCode: 'http_500', + httpStatusClass: '5xx', + progress: { processed_count: USER_DELETION_RESOURCE_BATCH_SIZE }, + }); + expect(await remainingSessionIds(user.id)).toHaveLength(1); + }); + + it('treats a successful DELETE that leaves the row as an identity mismatch', async () => { + const user = await insertTestUser(); + const sessionId = newSessionId('stuck'); + await insertSession(user.id, sessionId); + jest + .spyOn(globalThis, 'fetch') + .mockResolvedValue(new Response(JSON.stringify({ success: true }), { status: 200 })); + + const outcome = await handleCliV2Sessions({ + request: { user_id: user.id } as UserDeletionRequest, + step: runningStep(), + context: handlerContext(), + }); + + expect(outcome.kind).toBe('needs_attention'); + if (outcome.kind === 'needs_attention') { + expect(outcome.errorCode).toBe('session_identity_mismatch'); + } + expect(await remainingSessionIds(user.id)).toEqual([sessionId]); + }); + + it('handles authoritative absence without touching database', async () => { + const outcome = await handleCliV2Sessions({ + request: { + cloud_subject_resolution: UserDeletionCloudSubjectResolution.AuthoritativeAbsence, + } as UserDeletionRequest, + step: runningStep(), + context: handlerContext(), + }); + + expect(outcome).toEqual({ kind: 'not_applicable', errorCode: 'authoritative_absence' }); + }); +}); + +function runningStep(): UserDeletionStep { + return { + step_key: UserDeletionStepKey.CliV2Sessions, + status: UserDeletionStepStatus.Running, + progress_json: {}, + } as UserDeletionStep; +} + +function handlerContext(remainingMs: () => number = () => 60_000): DeletionHandlerContext { + return { + requestId: 'req-cli-v2', + stepKey: UserDeletionStepKey.CliV2Sessions, + claimToken: 'claim', + deadlineAt: Date.now() + remainingMs(), + remainingMs, + signal: new AbortController().signal, + }; +} + +function newSessionId(label: string): string { + return `ses_${label}${crypto.randomUUID().replaceAll('-', '')}`.slice(0, 30); +} + +async function insertSession(userId: string, sessionId: string, parentSessionId?: string) { + await db.insert(cli_sessions_v2).values({ + session_id: sessionId, + kilo_user_id: userId, + parent_session_id: parentSessionId, + created_on_platform: 'cli', + }); +} + +async function deleteSessionRow(userId: string, sessionId: string) { + await db + .delete(cli_sessions_v2) + .where( + and(eq(cli_sessions_v2.kilo_user_id, userId), eq(cli_sessions_v2.session_id, sessionId)) + ); +} + +async function remainingSessionIds(userId: string): Promise { + const rows = await db + .select({ session_id: cli_sessions_v2.session_id }) + .from(cli_sessions_v2) + .where(eq(cli_sessions_v2.kilo_user_id, userId)); + return rows.map(row => row.session_id).sort(); +} + +function sessionIdFromUrl(input: RequestInfo | URL): string { + const url = String(input); + if (!url.startsWith(INGEST_BASE)) { + throw new Error(`unexpected fetch ${url}`); + } + return decodeURIComponent(url.slice(INGEST_BASE.length)); +} + +function mockIngestDelete(userId: string, onDelete?: () => void) { + const deleted: string[] = []; + jest.spyOn(globalThis, 'fetch').mockImplementation(async input => { + const sessionId = sessionIdFromUrl(input); + await deleteSessionRow(userId, sessionId); + deleted.push(sessionId); + onDelete?.(); + return new Response(JSON.stringify({ success: true }), { status: 200 }); + }); + return deleted; +} diff --git a/apps/web/src/lib/user/deletion-queue/handlers/cli-v2.ts b/apps/web/src/lib/user/deletion-queue/handlers/cli-v2.ts index 144d51db31..44a0be9426 100644 --- a/apps/web/src/lib/user/deletion-queue/handlers/cli-v2.ts +++ b/apps/web/src/lib/user/deletion-queue/handlers/cli-v2.ts @@ -1,6 +1,7 @@ import { and, eq, notExists, sql } from 'drizzle-orm'; import { alias } from 'drizzle-orm/pg-core'; import { cli_sessions_v2 } from '@kilocode/db/schema'; +import type { UserDeletionTaskProgress } from '@kilocode/db/schema-types'; import { SESSION_INGEST_WORKER_URL } from '@/lib/config.server'; import { db } from '@/lib/drizzle'; import { generateInternalServiceToken } from '@/lib/tokens'; @@ -9,6 +10,10 @@ import { USER_DELETION_SESSION_INGEST_AUDIENCE, } from '@/lib/user/deletion-queue/deletion-constants'; import { userIdKeyedAbsenceOutcome } from '@/lib/user/deletion-queue/deletion-subject'; +import type { + DeletionHandlerContext, + DeletionHandlerOutcome, +} from '@/lib/user/deletion-queue/deletion-types'; import { classifyResponse, configurationMissing, @@ -65,6 +70,66 @@ async function anySessionExists(userId: string): Promise { return Boolean(row); } +type LeafDeleteResult = + | { kind: 'deleted' } + | { kind: 'conflict' } + | { kind: 'failed'; outcome: DeletionHandlerOutcome }; + +async function deleteLeafSession( + context: DeletionHandlerContext, + token: string, + userId: string, + sessionId: string +): Promise { + const url = `${SESSION_INGEST_WORKER_URL}/api/session/${encodeURIComponent(sessionId)}`; + const result = await deletionFetch(context, url, { + method: 'DELETE', + headers: { Authorization: `Bearer ${token}` }, + }); + if ('outcome' in result) return { kind: 'failed', outcome: result.outcome }; + + if (result.response.status === 409) return { kind: 'conflict' }; + if (!result.response.ok && result.response.status !== 404) { + return { kind: 'failed', outcome: classifyResponse(result.response) }; + } + + const stillPresent = await sessionExists(userId, sessionId); + if (stillPresent) { + return { + kind: 'failed', + outcome: { + kind: 'needs_attention', + errorCode: 'session_identity_mismatch', + resourceHmac: resourceHmac(sessionId), + }, + }; + } + if (result.response.status === 404) { + const body = await readJsonUnknown(result.response); + if (!isRecord(body) || body.cleanup !== 'done') { + return { + kind: 'failed', + outcome: { + kind: 'needs_attention', + errorCode: 'session_cleanup_unconfirmed', + resourceHmac: resourceHmac(sessionId), + }, + }; + } + } + return { kind: 'deleted' }; +} + +function withProgress( + outcome: DeletionHandlerOutcome, + progress: UserDeletionTaskProgress +): DeletionHandlerOutcome { + if (outcome.kind === 'not_applicable' || outcome.kind === 'manual_action_required') { + return outcome; + } + return { ...outcome, progress }; +} + export const handleCliV2Sessions: DeletionHandler = async ({ request, step, context }) => { const absence = userIdKeyedAbsenceOutcome(request); if (absence) return absence; @@ -73,66 +138,47 @@ export const handleCliV2Sessions: DeletionHandler = async ({ request, step, cont if (!SESSION_INGEST_WORKER_URL) return configurationMissing(); - const stop = continueIfLowTime(context, step.progress_json); - if (stop) return stop; - - const leaves = await loadLeafSessions(userId); - if (leaves.length === 0) { - if (await anySessionExists(userId)) { - return { kind: 'needs_attention', errorCode: 'cyclic_session_graph' }; - } - return (step.progress_json.processed_count ?? 0) === 0 - ? { kind: 'not_applicable' } - : { kind: 'succeeded', progress: step.progress_json }; - } - const token = generateInternalServiceToken(userId, { expiresIn: 5 * 60, audience: USER_DELETION_SESSION_INGEST_AUDIENCE, }); let progress = step.progress_json; - for (const leaf of leaves) { - const reserve = continueIfLowTime(context, progress); - if (reserve) return reserve; + while (true) { + const stop = continueIfLowTime(context, progress); + if (stop) return stop; - const url = `${SESSION_INGEST_WORKER_URL}/api/session/${encodeURIComponent(leaf.session_id)}`; - const result = await deletionFetch(context, url, { - method: 'DELETE', - headers: { Authorization: `Bearer ${token}` }, - }); - if ('outcome' in result) return result.outcome; - - if (result.response.status === 409) { - return { kind: 'continue', progress }; - } - if (!result.response.ok && result.response.status !== 404) { - return classifyResponse(result.response); + const leaves = await loadLeafSessions(userId); + if (leaves.length === 0) { + if (await anySessionExists(userId)) { + return { kind: 'needs_attention', errorCode: 'cyclic_session_graph', progress }; + } + return (progress.processed_count ?? 0) === 0 + ? { kind: 'not_applicable' } + : { kind: 'succeeded', progress }; } - const stillPresent = await sessionExists(userId, leaf.session_id); - if (stillPresent) { - return { - kind: 'needs_attention', - errorCode: 'session_identity_mismatch', - resourceHmac: resourceHmac(leaf.session_id), - }; - } - if (result.response.status === 404) { - const body = await readJsonUnknown(result.response); - if (!isRecord(body) || body.cleanup !== 'done') { - return { - kind: 'needs_attention', - errorCode: 'session_cleanup_unconfirmed', - resourceHmac: resourceHmac(leaf.session_id), - }; + const results = await Promise.all( + leaves.map(leaf => deleteLeafSession(context, token, userId, leaf.session_id)) + ); + + let firstFailure: DeletionHandlerOutcome | null = null; + let sawConflict = false; + for (const result of results) { + if (result.kind === 'deleted') { + progress = incrementProcessed(progress); + continue; } + if (result.kind === 'conflict') { + sawConflict = true; + continue; + } + if (!firstFailure) firstFailure = result.outcome; } - progress = incrementProcessed(progress); - } - if (await anySessionExists(userId)) { - return { kind: 'continue', progress }; + if (firstFailure) return withProgress(firstFailure, progress); + if (sawConflict) { + return { kind: 'continue', progress }; + } } - return { kind: 'succeeded', progress }; }; diff --git a/apps/web/src/lib/user/deletion-queue/handlers/usage-prompt-prefixes.test.ts b/apps/web/src/lib/user/deletion-queue/handlers/usage-prompt-prefixes.test.ts index 9cba47cb78..d3aae6c38e 100644 --- a/apps/web/src/lib/user/deletion-queue/handlers/usage-prompt-prefixes.test.ts +++ b/apps/web/src/lib/user/deletion-queue/handlers/usage-prompt-prefixes.test.ts @@ -88,6 +88,7 @@ describe('handleUsagePromptPrefixes', () => { expect(first.kind).toBe('continue'); if (first.kind !== 'continue') throw new Error('expected first page to continue'); expect(first.progress?.processed_count).toBe(USER_DELETION_USAGE_PREFIX_BATCH_SIZE); + expect(first.progress?.scanned_count).toBe(USER_DELETION_USAGE_PREFIX_BATCH_SIZE); expect(first.progress?.cursor).toContain('\t'); const firstPageIds = rows.slice(0, USER_DELETION_USAGE_PREFIX_BATCH_SIZE).map(row => row.id); @@ -115,6 +116,7 @@ describe('handleUsagePromptPrefixes', () => { expect(second.kind).toBe('succeeded'); if (second.kind !== 'succeeded') throw new Error('expected second page to succeed'); expect(second.progress?.processed_count).toBe(rows.length); + expect(second.progress?.scanned_count).toBe(rows.length); const remaining = await db .select({ user_prompt_prefix: microdollar_usage_metadata.user_prompt_prefix }) @@ -140,7 +142,56 @@ describe('handleUsagePromptPrefixes', () => { expect(outcome).toMatchObject({ kind: 'succeeded', - progress: { processed_count: 0 }, + progress: { processed_count: 0, scanned_count: 1 }, + }); + }); + + it('keeps scanning already-clean pages in one claim until a dirty page finishes', async () => { + const { user, request, step, claimToken } = await prepareRunningStep(); + await insertUsageRows(user.id, USER_DELETION_USAGE_PREFIX_BATCH_SIZE, null); + const dirtyId = await insertUsage(user.id, 'still dirty', null, 2); + + const outcome = await handleUsagePromptPrefixes({ + request, + step, + context: handlerContext(request.id, claimToken), + }); + + expect(outcome).toMatchObject({ + kind: 'succeeded', + progress: { + processed_count: 1, + scanned_count: USER_DELETION_USAGE_PREFIX_BATCH_SIZE + 1, + }, + }); + const metadata = await loadMetadata(dirtyId); + expect(metadata?.user_prompt_prefix).toBeNull(); + }); + + it('yields continue after clean pages when the time budget is gone', async () => { + const { user, request, step, claimToken } = await prepareRunningStep(); + await insertUsageRows(user.id, USER_DELETION_USAGE_PREFIX_BATCH_SIZE * 2, null); + + let remainingMs = 20_000; + const outcome = await handleUsagePromptPrefixes({ + request, + step, + context: { + ...handlerContext(request.id, claimToken), + remainingMs: () => { + const current = remainingMs; + remainingMs = 1; + return current; + }, + }, + }); + + expect(outcome).toMatchObject({ + kind: 'continue', + progress: { + processed_count: 0, + scanned_count: USER_DELETION_USAGE_PREFIX_BATCH_SIZE, + }, }); }); @@ -284,7 +335,11 @@ async function insertUsage( return id; } -async function insertUsageRows(userId: string, count: number) { +async function insertUsageRows( + userId: string, + count: number, + userPromptPrefix: string | null = 'private prompt' +) { const start = Date.parse('2020-01-01T00:00:00.000Z'); const rows = Array.from({ length: count }, (_, index) => ({ id: crypto.randomUUID(), @@ -309,7 +364,7 @@ async function insertUsageRows(userId: string, count: number) { rows.map(row => ({ id: row.id, message_id: `message-${row.id}`, - user_prompt_prefix: 'private prompt', + user_prompt_prefix: userPromptPrefix, max_tokens: 1, latency: 1, })) diff --git a/apps/web/src/lib/user/deletion-queue/handlers/usage-prompt-prefixes.ts b/apps/web/src/lib/user/deletion-queue/handlers/usage-prompt-prefixes.ts index 63578e26cd..e80084852f 100644 --- a/apps/web/src/lib/user/deletion-queue/handlers/usage-prompt-prefixes.ts +++ b/apps/web/src/lib/user/deletion-queue/handlers/usage-prompt-prefixes.ts @@ -16,8 +16,15 @@ const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/ type ParsedUsagePrefixProgress = { cursor: UsagePromptPrefixCursor | null; processedCount: number; + scannedCount: number; }; +function parseOptionalCount(value: unknown): number | null { + if (value === undefined) return 0; + if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 0) return null; + return value; +} + function parseUsagePrefixProgress( progress: UserDeletionTaskProgress ): ParsedUsagePrefixProgress | null { @@ -27,20 +34,16 @@ function parseUsagePrefixProgress( const rawProgress = progress as { processed_count?: unknown; + scanned_count?: unknown; cursor?: unknown; }; - const processedCount = rawProgress.processed_count; - if (processedCount !== undefined && typeof processedCount !== 'number') return null; - if ( - processedCount !== undefined && - (!Number.isSafeInteger(processedCount) || processedCount < 0) - ) { - return null; - } + const processedCount = parseOptionalCount(rawProgress.processed_count); + const scannedCount = parseOptionalCount(rawProgress.scanned_count); + if (processedCount === null || scannedCount === null) return null; const rawCursor = rawProgress.cursor; if (rawCursor === undefined || rawCursor === '') { - return { cursor: null, processedCount: processedCount ?? 0 }; + return { cursor: null, processedCount, scannedCount }; } if (typeof rawCursor !== 'string') return null; @@ -59,7 +62,8 @@ function parseUsagePrefixProgress( return { cursor: { createdAt, id }, - processedCount: processedCount ?? 0, + processedCount, + scannedCount, }; } @@ -93,89 +97,106 @@ export const handleUsagePromptPrefixes: DeletionHandler = async ({ request, step const userId = request.user_id; if (!userId) return { kind: 'needs_attention', errorCode: 'legacy_identity_unresolved' }; - const parsedProgress = parseUsagePrefixProgress(step.progress_json); - if (!parsedProgress) { + const initialProgress = parseUsagePrefixProgress(step.progress_json); + if (!initialProgress) { return { kind: 'needs_attention', errorCode: 'usage_prefix_progress_invalid' }; } - const stop = continueIfLowTime(context, step.progress_json); - if (stop) return stop; - - try { - const result = await db.transaction(async tx => { - await tx.execute( - sql.raw(`SET LOCAL statement_timeout = ${USER_DELETION_USAGE_PREFIX_STATEMENT_TIMEOUT_MS}`) - ); - - const [lockedStep] = await tx - .select({ id: user_deletion_steps.id }) - .from(user_deletion_steps) - .where( - and( - eq(user_deletion_steps.request_id, request.id), - eq(user_deletion_steps.step_key, context.stepKey), - eq(user_deletion_steps.claim_token, context.claimToken), - eq(user_deletion_steps.status, UserDeletionStepStatus.Running) + let parsedProgress = initialProgress; + let progress: UserDeletionTaskProgress = step.progress_json; + + while (true) { + const stop = continueIfLowTime(context, progress); + if (stop) return stop; + + try { + const result = await db.transaction(async tx => { + await tx.execute( + sql.raw( + `SET LOCAL statement_timeout = ${USER_DELETION_USAGE_PREFIX_STATEMENT_TIMEOUT_MS}` ) - ) - .for('update'); - if (!lockedStep) throw new UsagePrefixClaimLostError(); - - const page = await scrubUsagePromptPrefixesPage( - tx, - userId, - parsedProgress.cursor, - USER_DELETION_USAGE_PREFIX_BATCH_SIZE - ); - if (page.pageSize === 0) return { page, progress: step.progress_json }; - - const progress: UserDeletionTaskProgress = { - processed_count: parsedProgress.processedCount + page.updatedCount, - cursor: page.lastCursor ? encodeCursor(page.lastCursor) : undefined, - }; - const checkpoint = await tx - .update(user_deletion_steps) - .set({ progress_json: progress }) - .where(eq(user_deletion_steps.id, lockedStep.id)) - .returning({ id: user_deletion_steps.id }); - if (checkpoint.length !== 1) throw new UsagePrefixClaimLostError(); - - await tx - .update(user_deletion_requests) - .set({ last_progress_at: sql`now()` }) - .where(eq(user_deletion_requests.id, request.id)); - - return { page, progress }; - }); - - if (result.page.pageSize === 0) { - return parsedProgress.processedCount === 0 - ? { kind: 'not_applicable' } - : { kind: 'succeeded', progress: result.progress }; - } - if (result.page.pageSize < USER_DELETION_USAGE_PREFIX_BATCH_SIZE) { - return { kind: 'succeeded', progress: result.progress }; - } - return { kind: 'continue', progress: result.progress }; - } catch (error) { - if (error instanceof UsagePrefixClaimLostError) { - return { kind: 'retry', errorCode: 'claim_lost', httpStatusClass: 'error' }; - } - const code = postgresErrorCode(error); - if (code === '57014') { - return { - kind: 'retry', - errorCode: 'usage_prefix_page_timeout', - httpStatusClass: 'error', - }; - } - if (code === '40001' || code === '40P01') { - return { - kind: 'retry', - errorCode: 'usage_prefix_page_failed', - httpStatusClass: 'error', + ); + + const [lockedStep] = await tx + .select({ id: user_deletion_steps.id }) + .from(user_deletion_steps) + .where( + and( + eq(user_deletion_steps.request_id, request.id), + eq(user_deletion_steps.step_key, context.stepKey), + eq(user_deletion_steps.claim_token, context.claimToken), + eq(user_deletion_steps.status, UserDeletionStepStatus.Running) + ) + ) + .for('update'); + if (!lockedStep) throw new UsagePrefixClaimLostError(); + + const page = await scrubUsagePromptPrefixesPage( + tx, + userId, + parsedProgress.cursor, + USER_DELETION_USAGE_PREFIX_BATCH_SIZE + ); + if (page.pageSize === 0) return { page, progress }; + + const nextProgress: UserDeletionTaskProgress = { + processed_count: parsedProgress.processedCount + page.updatedCount, + scanned_count: parsedProgress.scannedCount + page.pageSize, + cursor: page.lastCursor ? encodeCursor(page.lastCursor) : undefined, + }; + const checkpoint = await tx + .update(user_deletion_steps) + .set({ progress_json: nextProgress }) + .where(eq(user_deletion_steps.id, lockedStep.id)) + .returning({ id: user_deletion_steps.id }); + if (checkpoint.length !== 1) throw new UsagePrefixClaimLostError(); + + await tx + .update(user_deletion_requests) + .set({ last_progress_at: sql`now()` }) + .where(eq(user_deletion_requests.id, request.id)); + + return { page, progress: nextProgress }; + }); + + if (result.page.pageSize === 0) { + return parsedProgress.processedCount === 0 && parsedProgress.scannedCount === 0 + ? { kind: 'not_applicable' } + : { kind: 'succeeded', progress }; + } + + progress = result.progress; + parsedProgress = { + cursor: result.page.lastCursor, + processedCount: parsedProgress.processedCount + result.page.updatedCount, + scannedCount: parsedProgress.scannedCount + result.page.pageSize, }; + if (result.page.pageSize < USER_DELETION_USAGE_PREFIX_BATCH_SIZE) { + return { kind: 'succeeded', progress }; + } + if (result.page.updatedCount > 0) { + return { kind: 'continue', progress }; + } + } catch (error) { + if (error instanceof UsagePrefixClaimLostError) { + return { kind: 'retry', errorCode: 'claim_lost', httpStatusClass: 'error' }; + } + const code = postgresErrorCode(error); + if (code === '57014') { + return { + kind: 'retry', + errorCode: 'usage_prefix_page_timeout', + httpStatusClass: 'error', + }; + } + if (code === '40001' || code === '40P01') { + return { + kind: 'retry', + errorCode: 'usage_prefix_page_failed', + httpStatusClass: 'error', + }; + } + throw error; } - throw error; } }; diff --git a/apps/web/src/routers/admin/user-deletion-queue-router.ts b/apps/web/src/routers/admin/user-deletion-queue-router.ts index a14c52972c..f6193fa1cc 100644 --- a/apps/web/src/routers/admin/user-deletion-queue-router.ts +++ b/apps/web/src/routers/admin/user-deletion-queue-router.ts @@ -155,6 +155,7 @@ function isStaleRequest(lastProgressAt: string, asOf: string): boolean { function taskProgress(step: UserDeletionStep) { return { processedCount: step.progress_json.processed_count ?? 0, + scannedCount: step.progress_json.scanned_count ?? 0, pageOffset: step.progress_json.page_offset ?? null, cleanPass: step.progress_json.clean_pass ?? false, }; @@ -170,6 +171,7 @@ function serializeTaskSummary(step: UserDeletionStep) { lifetimeAttemptCount: step.lifetime_attempt_count, availableAt: iso(step.available_at), processedCount: progress.processedCount, + scannedCount: progress.scannedCount, pageOffset: progress.pageOffset, cleanPass: progress.cleanPass, rateLimitedSince: nullableIso(step.rate_limited_since), @@ -207,6 +209,7 @@ function serializeRequest(request: UserDeletionRequest, tasks: UserDeletionStep[ lifetimeAttemptCount: 0, availableAt: iso(request.created_at), processedCount: 0, + scannedCount: 0, pageOffset: null, cleanPass: false, rateLimitedSince: null, @@ -229,6 +232,7 @@ function serializeActivity(row: UserDeletionActivity) { details: { durationMs: row.details_json.duration_ms ?? null, processedCount: row.details_json.processed_count ?? null, + scannedCount: row.details_json.scanned_count ?? null, httpStatusClass: row.details_json.http_status_class ?? null, retryAt: row.details_json.retry_at ?? null, errorCode: row.details_json.error_code ?? null, diff --git a/packages/db/src/schema-types.ts b/packages/db/src/schema-types.ts index 7b3299fabf..384e50b22c 100644 --- a/packages/db/src/schema-types.ts +++ b/packages/db/src/schema-types.ts @@ -569,6 +569,7 @@ export type UserDeletionPylonReplyState = export type UserDeletionTaskProgress = { processed_count?: number; + scanned_count?: number; page_offset?: number; cursor?: string; clean_pass?: boolean; @@ -597,6 +598,7 @@ export type UserDeletionAuditDetails = { export type UserDeletionActivityDetails = { duration_ms?: number; processed_count?: number; + scanned_count?: number; http_status_class?: string; retry_at?: string; error_code?: string;