diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 396fb1e1685..8a12ade46b8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -371,6 +371,23 @@ jobs: if: needs.changes.outputs.cloud_agent_next == 'true' runs-on: ${{ vars.RUNNER_LARGE_LABEL || 'ubuntu-24.04-8core' }} timeout-minutes: 15 + + env: + POSTGRES_URL: postgresql://postgres:postgres@localhost:5432/postgres + + services: + postgres: + image: pgvector/pgvector:pg16 + env: + POSTGRES_HOST_AUTH_METHOD: trust + options: >- + --health-cmd pg_isready + --health-interval 10s + --health-timeout 5s + --health-retries 5 + ports: + - 5432:5432 + steps: - uses: useblacksmith/checkout@41cdeedae8edb2e684ba22896a5fd2a3cb85db6b # v1 with: @@ -401,6 +418,9 @@ jobs: - name: Run cloud-agent-next tests run: pnpm --filter cloud-agent-next test:all + - name: Run Cloud Agent Postgres aggregate tests + run: NODE_ENV=development pnpm drizzle migrate && pnpm --filter cloud-agent-next test:postgres + workspace-tests: needs: [changes, typecheck, lint, format-check, drizzle-check] if: ${{ needs.changes.outputs.workspace_matrix != '[]' }} diff --git a/services/cloud-agent-next/package.json b/services/cloud-agent-next/package.json index 8d47332bd04..f2598ee9d93 100644 --- a/services/cloud-agent-next/package.json +++ b/services/cloud-agent-next/package.json @@ -16,11 +16,12 @@ "prestart": "pnpm run build:wrapper", "start": "scripts/dev-with-docker-proxy.sh --no-proxy", "types": "wrangler types", - "lint": "pnpm -w exec oxlint --config .oxlintrc.json services/cloud-agent-next/src services/cloud-agent-next/scripts services/cloud-agent-next/wrapper/src", - "format": "oxfmt src scripts", - "format:check": "oxfmt --list-different src scripts", + "lint": "pnpm -w exec oxlint --config .oxlintrc.json services/cloud-agent-next/src services/cloud-agent-next/scripts services/cloud-agent-next/wrapper/src services/cloud-agent-next/test/postgres", + "format": "oxfmt src scripts test/postgres", + "format:check": "oxfmt --list-different src scripts test/postgres", "build:wrapper": "bun run --cwd wrapper build", "test": "vitest run", + "test:postgres": "vitest run --config vitest.postgres.config.ts", "test:watch": "vitest", "test:integration": "vitest run --config vitest.workers.config.ts", "test:integration:watch": "vitest --config vitest.workers.config.ts", diff --git a/services/cloud-agent-next/src/server.test.ts b/services/cloud-agent-next/src/server.test.ts index 8914405ec22..5a34dd18fbe 100644 --- a/services/cloud-agent-next/src/server.test.ts +++ b/services/cloud-agent-next/src/server.test.ts @@ -1,3 +1,7 @@ +import fs from 'node:fs'; +import path from 'node:path'; + +import { parse } from 'jsonc-parser'; import { beforeEach, describe, expect, it, vi } from 'vitest'; import jwt from 'jsonwebtoken'; import { VERCEL_SANDBOX_UNAVAILABLE_MESSAGE } from './agent-sandbox/vercel/vercel-agent-sandbox.js'; @@ -17,21 +21,25 @@ const { getRunningTerminalClientMock, consumeCloudAgentReportBatchMock, removeExpiredCloudAgentReportDataMock, + runCloudAgentOutcomeCollectionMock, requireCurrentSessionAccessMock, getPgDbMock, + loggerWarnMock, } = vi.hoisted(() => ({ getRunningTerminalClientMock: vi.fn(), consumeCloudAgentReportBatchMock: vi.fn().mockResolvedValue(undefined), removeExpiredCloudAgentReportDataMock: vi.fn().mockResolvedValue(undefined), + runCloudAgentOutcomeCollectionMock: vi.fn().mockResolvedValue(undefined), requireCurrentSessionAccessMock: vi.fn(), getPgDbMock: vi.fn(), + loggerWarnMock: vi.fn(), })); vi.mock('./logger.js', () => { const logger = { setTags: vi.fn(), info: vi.fn(), - warn: vi.fn(), + warn: loggerWarnMock, error: vi.fn(), withFields: vi.fn(), }; @@ -80,6 +88,10 @@ vi.mock('./telemetry/report-consumer.js', () => ({ removeExpiredCloudAgentReportData: removeExpiredCloudAgentReportDataMock, })); +vi.mock('./telemetry/outcome-aggregate.js', () => ({ + runCloudAgentOutcomeCollection: runCloudAgentOutcomeCollectionMock, +})); + vi.mock('./middleware/auth.js', () => ({ authMiddleware: vi.fn(), })); @@ -129,7 +141,11 @@ vi.mock('@kilocode/db/client', () => ({ }), })); -const { default: worker } = await import('./server.js'); +const { + default: worker, + REPORT_RETENTION_CRON, + OUTCOME_AGGREGATE_CRON, +} = await import('./server.js'); const secret = 'test-secret'; @@ -265,6 +281,8 @@ beforeEach(() => { getRunningTerminalClientMock.mockReset(); consumeCloudAgentReportBatchMock.mockClear(); removeExpiredCloudAgentReportDataMock.mockClear(); + runCloudAgentOutcomeCollectionMock.mockClear(); + loggerWarnMock.mockClear(); getPgDbMock.mockReset(); requireCurrentSessionAccessMock.mockReset().mockResolvedValue({ kiloSessionId: 'ses_12345678901234567890123456', @@ -365,12 +383,54 @@ describe('server background reporting', () => { expect(consumeCloudAgentReportBatchMock).toHaveBeenCalledWith(batch, env); }); - it('runs reporting retention cleanup from the scheduled handler', async () => { + it('runs only reporting retention cleanup on the daily cron', async () => { const env = createEnv(); - await worker.scheduled({} as ScheduledController, env as unknown as Env); + await worker.scheduled( + { cron: REPORT_RETENTION_CRON } as ScheduledController, + env as unknown as Env + ); + expect(removeExpiredCloudAgentReportDataMock).toHaveBeenCalledTimes(1); expect(removeExpiredCloudAgentReportDataMock).toHaveBeenCalledWith(env); + expect(runCloudAgentOutcomeCollectionMock).not.toHaveBeenCalled(); + }); + + it('runs only the outcome collection on the 3-minute cron', async () => { + const env = createEnv(); + + await worker.scheduled( + { cron: OUTCOME_AGGREGATE_CRON } as ScheduledController, + env as unknown as Env + ); + + expect(runCloudAgentOutcomeCollectionMock).toHaveBeenCalledTimes(1); + expect(runCloudAgentOutcomeCollectionMock).toHaveBeenCalledWith(env); + expect(removeExpiredCloudAgentReportDataMock).not.toHaveBeenCalled(); + }); + + it('runs neither branch for an unrecognized cron and warns', async () => { + const env = createEnv(); + + await worker.scheduled({ cron: '0 0 * * 0' } as ScheduledController, env as unknown as Env); + + expect(removeExpiredCloudAgentReportDataMock).not.toHaveBeenCalled(); + expect(runCloudAgentOutcomeCollectionMock).not.toHaveBeenCalled(); + expect(loggerWarnMock).toHaveBeenCalledWith( + 'Cloud Agent scheduled handler received an unrecognized cron', + { cron: '0 0 * * 0' } + ); + }); + + it('keeps the deployed cron triggers in sync with the dispatcher constants', () => { + const config = parse(fs.readFileSync(path.join(process.cwd(), 'wrangler.jsonc'), 'utf8')) as { + triggers?: { crons?: string[] }; + env?: { dev?: { triggers?: { crons?: string[] } } }; + }; + const expected = [REPORT_RETENTION_CRON, OUTCOME_AGGREGATE_CRON].sort(); + + expect((config.triggers?.crons ?? []).slice().sort()).toEqual(expected); + expect((config.env?.dev?.triggers?.crons ?? []).slice().sort()).toEqual(expected); }); }); diff --git a/services/cloud-agent-next/src/server.ts b/services/cloud-agent-next/src/server.ts index 77bdc8756e2..7a782a1b66c 100644 --- a/services/cloud-agent-next/src/server.ts +++ b/services/cloud-agent-next/src/server.ts @@ -23,6 +23,7 @@ import { consumeCloudAgentReportBatch, removeExpiredCloudAgentReportData, } from './telemetry/report-consumer.js'; +import { runCloudAgentOutcomeCollection } from './telemetry/outcome-aggregate.js'; import { authMiddleware } from './middleware/auth.js'; import { balanceMiddleware } from './middleware/balance.js'; import { resolveTerminalWrapperClient } from './terminal/access.js'; @@ -1089,6 +1090,9 @@ app.use( app.notFound(createNotFoundHandler()); app.onError(createErrorHandler(logger, { includeMessage: false })); +export const REPORT_RETENTION_CRON = '17 2 * * *'; +export const OUTCOME_AGGREGATE_CRON = '*/3 * * * *'; + export default { fetch(request: Request, env: Env, ctx: ExecutionContext): Response | Promise { const url = new URL(request.url); @@ -1112,8 +1116,18 @@ export default { logger.warn(`Received message from unexpected queue: ${batch.queue}`); }, - async scheduled(_controller: ScheduledController, env: Env): Promise { - await removeExpiredCloudAgentReportData(env); + async scheduled(controller: ScheduledController, env: Env): Promise { + if (controller.cron === REPORT_RETENTION_CRON) { + await removeExpiredCloudAgentReportData(env); + return; + } + if (controller.cron === OUTCOME_AGGREGATE_CRON) { + await runCloudAgentOutcomeCollection(env); + return; + } + logger.warn('Cloud Agent scheduled handler received an unrecognized cron', { + cron: controller.cron, + }); }, }; diff --git a/services/cloud-agent-next/src/session-plane.ts b/services/cloud-agent-next/src/session-plane.ts index 7bcbc7db2ce..86bfe657278 100644 --- a/services/cloud-agent-next/src/session-plane.ts +++ b/services/cloud-agent-next/src/session-plane.ts @@ -2,12 +2,14 @@ import type { SessionId } from './types.js'; export type SessionPlane = 'legacy' | 'control'; +export const CONTROL_PLANE_SESSION_PREFIX = 'workspace_'; + export type ControlPlaneOwnerEnv = { CONTROL_PLANE_IDS?: string; }; export function sessionPlaneFromId(sessionId: string): SessionPlane { - return sessionId.startsWith('workspace_') ? 'control' : 'legacy'; + return sessionId.startsWith(CONTROL_PLANE_SESSION_PREFIX) ? 'control' : 'legacy'; } export function sessionSupportsTerminal(sessionId: string): boolean { @@ -17,7 +19,7 @@ export function sessionSupportsTerminal(sessionId: string): boolean { export function generateSessionId(plane: SessionPlane = 'legacy'): SessionId { const id = crypto.randomUUID(); - return plane === 'control' ? `workspace_${id}` : `agent_${id}`; + return plane === 'control' ? `${CONTROL_PLANE_SESSION_PREFIX}${id}` : `agent_${id}`; } export type SessionCreateOrigin = { diff --git a/services/cloud-agent-next/src/telemetry/outcome-aggregate.test.ts b/services/cloud-agent-next/src/telemetry/outcome-aggregate.test.ts new file mode 100644 index 00000000000..5d16ff6e125 --- /dev/null +++ b/services/cloud-agent-next/src/telemetry/outcome-aggregate.test.ts @@ -0,0 +1,374 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import type { SQL } from 'drizzle-orm'; +import { eq } from 'drizzle-orm'; +import { getWorkerDb } from '@kilocode/db/client'; +import { cloud_agent_session_runs, cloud_agent_sessions } from '@kilocode/db/schema'; + +const { getPgDbMock, loggerMock } = vi.hoisted(() => ({ + getPgDbMock: vi.fn(), + loggerMock: { + withFields: vi.fn(), + info: vi.fn(), + error: vi.fn(), + warn: vi.fn(), + }, +})); + +vi.mock('../db/pg.js', () => ({ getPgDb: getPgDbMock })); +vi.mock('../logger.js', () => ({ logger: loggerMock })); + +import { + AGGREGATE_METRIC, + OUTCOME_AGGREGATE_LIMITATIONS, + OUTCOME_WINDOW_MINUTES, + assembleGenerationAggregates, + readOutcomeAggregate, + runCloudAgentOutcomeCollection, + type OutcomeGeneration, + type OutcomeRole, + type OutcomeWindow, +} from './outcome-aggregate.js'; + +const window: OutcomeWindow = { + windowMinutes: 5, + start: '2026-02-01T00:05:00.000Z', + end: '2026-02-01T00:10:00.000Z', +}; +const retentionCutoffIso = '2025-11-03T00:10:00.000Z'; + +type QueryResult = unknown[] | Error; + +function makeDb(results: QueryResult[]) { + const selects: Array> = []; + let transactionConfig: unknown; + const tx = { + select(fields: Record) { + selects.push(fields); + const result = results.shift(); + const chain = { + from: () => chain, + innerJoin: () => chain, + where: () => chain, + groupBy: () => chain, + then(resolve: (value: unknown) => unknown, reject: (reason: unknown) => unknown): unknown { + if (result instanceof Error) return reject(result); + return resolve(result ?? []); + }, + }; + return chain; + }, + }; + const db = { + transaction: vi.fn( + async (operation: (transaction: typeof tx) => Promise, config: unknown) => { + transactionConfig = config; + return operation(tx); + } + ), + }; + return { + db, + selects, + config: () => transactionConfig, + }; +} + +function runRow(input: { + generation: OutcomeGeneration; + role: OutcomeRole; + status: 'completed' | 'failed' | 'interrupted'; + responsibility?: string; + runCount: number; + failureStage?: string; + failureCode?: string; +}) { + return { + generation: input.generation, + role: input.role, + status: input.status, + responsibility: input.responsibility ?? 'unknown', + failureStage: input.failureStage ?? 'unknown', + failureCode: input.failureCode ?? 'unclassified', + runCount: input.runCount, + }; +} + +const usageDb = () => getWorkerDb('postgres://unused:unused@localhost:0/unused'); + +beforeEach(() => { + vi.clearAllMocks(); + loggerMock.withFields.mockReturnValue(loggerMock); +}); + +describe('cloud agent outcome aggregate assembly', () => { + it('reports both generations and both roles as zero-row results with null shares', () => { + const generations = assembleGenerationAggregates([], [], []); + + expect(generations.map(entry => entry.generation)).toEqual(['legacy', 'control']); + for (const entry of generations) { + expect(entry.runRowsObserved).toBe(false); + expect(entry.initial.settled).toBe(0); + expect(entry.followUp.settled).toBe(0); + expect(entry.totals.settled).toBe(0); + expect(entry.totals.allFailed).toBe(0); + expect(entry.totals.platformFailureShare).toBeNull(); + expect(entry.totals.unknownClassificationShare).toBeNull(); + expect(entry.distinctPlatformAffectedSessions).toBe(0); + expect(entry.distinctUnknownAffectedSessions).toBe(0); + expect(entry.sessionSetupFailures).toEqual([]); + expect(entry.sessionSetupFailureCount).toBe(0); + } + }); + + it('treats an unknown responsibility bucket as unknownFailed and computes shares', () => { + const generations = assembleGenerationAggregates( + [ + runRow({ generation: 'legacy', role: 'initial', status: 'completed', runCount: 3 }), + runRow({ + generation: 'legacy', + role: 'initial', + status: 'failed', + responsibility: 'unknown', + runCount: 1, + failureStage: 'agent_activity', + failureCode: 'assistant_error', + }), + runRow({ + generation: 'legacy', + role: 'initial', + status: 'failed', + responsibility: 'platform', + runCount: 1, + failureStage: 'pre_dispatch', + failureCode: 'sandbox_connect_failed', + }), + runRow({ + generation: 'legacy', + role: 'initial', + status: 'interrupted', + runCount: 4, + }), + ], + [ + { + generation: 'legacy', + distinctPlatformAffectedSessions: 1, + distinctUnknownAffectedSessions: 1, + }, + ], + [] + ); + + const legacy = generations[0]; + expect(legacy.runRowsObserved).toBe(true); + expect(legacy.initial.completed).toBe(3); + expect(legacy.initial.unknownFailed).toBe(1); + expect(legacy.initial.platformFailed).toBe(1); + expect(legacy.initial.interrupted).toBe(4); + expect(legacy.initial.allFailed).toBe(2); + expect(legacy.initial.settled).toBe(5); + expect(legacy.initial.platformFailureShare).toBeCloseTo(1 / 5); + expect(legacy.initial.unknownClassificationShare).toBeCloseTo(1 / 2); + expect(legacy.initial.failureStages).toEqual([ + { stage: 'agent_activity', count: 1 }, + { stage: 'pre_dispatch', count: 1 }, + ]); + expect(legacy.initial.failureStageCodes).toEqual([ + { stage: 'agent_activity', code: 'assistant_error', count: 1 }, + { stage: 'pre_dispatch', code: 'sandbox_connect_failed', count: 1 }, + ]); + expect(legacy.distinctPlatformAffectedSessions).toBe(1); + expect(legacy.distinctUnknownAffectedSessions).toBe(1); + }); + + it('separates initial runs from follow-up runs and recomputes totals from summed counts', () => { + const generations = assembleGenerationAggregates( + [ + runRow({ + generation: 'control', + role: 'initial', + status: 'failed', + responsibility: 'platform', + runCount: 1, + failureStage: 'pre_dispatch', + failureCode: 'sandbox_connect_failed', + }), + runRow({ + generation: 'control', + role: 'follow_up', + status: 'failed', + responsibility: 'platform', + runCount: 2, + failureStage: 'pre_dispatch', + failureCode: 'sandbox_connect_failed', + }), + runRow({ generation: 'control', role: 'follow_up', status: 'completed', runCount: 1 }), + ], + [], + [] + ); + + const control = generations[1]; + expect(control.generation).toBe('control'); + expect(control.initial.platformFailed).toBe(1); + expect(control.initial.completed).toBe(0); + expect(control.followUp.platformFailed).toBe(2); + expect(control.followUp.completed).toBe(1); + expect(control.totals.platformFailed).toBe(3); + expect(control.totals.completed).toBe(1); + expect(control.totals.settled).toBe(4); + expect(control.totals.platformFailureShare).toBeCloseTo(3 / 4); + expect(control.totals.failureStages).toEqual([{ stage: 'pre_dispatch', count: 3 }]); + }); + + it('reports session setup failures separately from settled runs', () => { + const generations = assembleGenerationAggregates( + [], + [], + [ + { + generation: 'legacy', + failureStage: 'initial_admission', + failureCode: 'initial_queue_full', + failureCount: 2, + }, + { + generation: 'legacy', + failureStage: 'registration', + failureCode: 'do_registration_rejected', + failureCount: 1, + }, + ] + ); + + const legacy = generations[0]; + expect(legacy.totals.settled).toBe(0); + expect(legacy.sessionSetupFailures).toEqual([ + { stage: 'initial_admission', code: 'initial_queue_full', count: 2 }, + { stage: 'registration', code: 'do_registration_rejected', count: 1 }, + ]); + expect(legacy.sessionSetupFailureCount).toBe(3); + }); +}); + +describe('cloud agent outcome aggregate wiring', () => { + it('runs all three queries sequentially on one read-only repeatable-read transaction', async () => { + const fake = makeDb([ + [runRow({ generation: 'legacy', role: 'initial', status: 'completed', runCount: 1 })], + [], + [], + ]); + + const generations = await readOutcomeAggregate(fake.db as never, { + window, + retentionCutoff: retentionCutoffIso, + }); + + expect(fake.db.transaction).toHaveBeenCalledTimes(1); + expect(fake.config()).toEqual({ isolationLevel: 'repeatable read', accessMode: 'read only' }); + expect(fake.selects).toHaveLength(3); + expect(generations).toHaveLength(2); + }); + + it('derives the generation prefix, initial role and unknown bucket from exact SQL predicates', async () => { + const fake = makeDb([[], [], []]); + await readOutcomeAggregate(fake.db as never, { window, retentionCutoff: retentionCutoffIso }); + + const [runSelect, distinctSelect] = fake.selects; + const render = (expression: SQL) => + usageDb() + .select({ expression }) + .from(cloud_agent_session_runs) + .innerJoin( + cloud_agent_sessions, + eq( + cloud_agent_sessions.cloud_agent_session_id, + cloud_agent_session_runs.cloud_agent_session_id + ) + ) + .toSQL(); + + const role = render(runSelect.role as SQL); + expect(role.sql).toMatch( + /"cloud_agent_session_runs"\."message_id" = "cloud_agent_sessions"\."initial_message_id"/ + ); + + const generation = render(runSelect.generation as SQL); + expect(generation.sql).toMatch( + /left\("cloud_agent_session_runs"\."cloud_agent_session_id", \$1\) = \$2/ + ); + expect(generation.params).toEqual([10, 'workspace_']); + + const responsibility = render(runSelect.responsibility as SQL); + expect(responsibility.sql).toContain('"failure_responsibility" is null'); + expect(responsibility.sql).toContain(`"failure_responsibility" not in ('platform', 'user')`); + + const distinctUnknown = render(distinctSelect.distinctUnknownAffectedSessions as SQL); + expect(distinctUnknown.sql).toContain( + 'count(distinct "cloud_agent_session_runs"."cloud_agent_session_id") filter (where' + ); + expect(distinctUnknown.sql).toContain('"failure_responsibility" is null'); + expect(distinctUnknown.sql).toContain(`"failure_responsibility" not in ('platform', 'user')`); + expect(distinctUnknown.sql).toContain('::int'); + + const distinctPlatform = render(distinctSelect.distinctPlatformAffectedSessions as SQL); + expect(distinctPlatform.sql).toContain( + 'count(distinct "cloud_agent_session_runs"."cloud_agent_session_id") filter (where "cloud_agent_session_runs"."failure_responsibility" = \'platform\')' + ); + expect(distinctPlatform.sql).toContain('::int'); + }); + + it('rejects when a later query fails after an earlier query succeeded', async () => { + const fake = makeDb([ + [runRow({ generation: 'legacy', role: 'initial', status: 'completed', runCount: 1 })], + [], + new Error('outcome q3 failed'), + ]); + + await expect( + readOutcomeAggregate(fake.db as never, { window, retentionCutoff: retentionCutoffIso }) + ).rejects.toThrow('outcome q3 failed'); + expect(fake.selects).toHaveLength(3); + }); +}); + +describe('runCloudAgentOutcomeCollection emission', () => { + it('emits one complete record per window and no failed record', async () => { + getPgDbMock.mockReturnValue(makeDb([[], [], [], [], [], [], [], [], []]).db); + + await runCloudAgentOutcomeCollection({} as never, new Date('2026-02-01T00:10:00.000Z')); + + expect(loggerMock.info).toHaveBeenCalledTimes(OUTCOME_WINDOW_MINUTES.length); + expect(loggerMock.error).not.toHaveBeenCalled(); + const record = loggerMock.withFields.mock.calls[0]?.[0] as Record; + expect(record.metric).toBe(AGGREGATE_METRIC); + expect(record.logTag).toBe(AGGREGATE_METRIC); + expect(record.collectionStatus).toBe('complete'); + expect(record.limitations).toEqual([...OUTCOME_AGGREGATE_LIMITATIONS]); + expect((record.generations as unknown[]).length).toBe(2); + }); + + it('emits exactly one failed record with no counts when a query fails', async () => { + getPgDbMock.mockReturnValue( + makeDb([ + [runRow({ generation: 'legacy', role: 'initial', status: 'completed', runCount: 1 })], + [], + new Error('outcome q3 failed'), + ]).db + ); + + await expect( + runCloudAgentOutcomeCollection({} as never, new Date('2026-02-01T00:10:00.000Z')) + ).resolves.toBeUndefined(); + + expect(loggerMock.error).toHaveBeenCalledTimes(1); + expect(loggerMock.info).not.toHaveBeenCalled(); + const record = loggerMock.withFields.mock.calls.at(-1)?.[0] as Record; + expect(record.collectionStatus).toBe('failed'); + expect(record.failureKind).toBe('db_query_failed'); + expect(record.generations).toBeUndefined(); + expect(record.windowMinutes).toBe(OUTCOME_WINDOW_MINUTES[0]); + expect(record.collectionTimestamp).toBe('2026-02-01T00:10:00.000Z'); + expect(record.windowEnd).toBe('2026-02-01T00:08:00.000Z'); + }); +}); diff --git a/services/cloud-agent-next/src/telemetry/outcome-aggregate.ts b/services/cloud-agent-next/src/telemetry/outcome-aggregate.ts new file mode 100644 index 00000000000..baaf04692b8 --- /dev/null +++ b/services/cloud-agent-next/src/telemetry/outcome-aggregate.ts @@ -0,0 +1,388 @@ +import { and, eq, gt, gte, lt, sql, type AnyColumn, type SQL } from 'drizzle-orm'; +import type { WorkerDb } from '@kilocode/db/client'; +import { + cloud_agent_session_runs, + cloud_agent_sessions, + type CloudAgentSessionRunStatus, +} from '@kilocode/db/schema'; +import { getPgDb } from '../db/pg.js'; +import type { Env } from '../types.js'; +import { logger } from '../logger.js'; +import { CONTROL_PLANE_SESSION_PREFIX } from '../session-plane.js'; +import { CLOUD_AGENT_REPORT_RETENTION_DAYS, retentionCutoff } from './report-store.js'; + +export const OUTCOME_WINDOW_MINUTES = [5, 15, 60] as const; +// Provisional: not yet measured against production reporting lag. +export const REPORTING_DELAY_ALLOWANCE_MS = 2 * 60 * 1000; +export const COVERAGE_LABEL = `retained_reporting_population_${CLOUD_AGENT_REPORT_RETENTION_DAYS}d`; +export const AGGREGATE_CONTRACT_VERSION = 1; +export const AGGREGATE_METRIC = 'cloud_agent_outcome_aggregate'; +export const EXPECTED_GENERATIONS = ['legacy', 'control'] as const; +export const OUTCOME_AGGREGATE_LIMITATIONS = [ + 'reported_categories_only', + 'pre_dispatch_not_proof_of_model_reach', + 'post_dispatch_no_activity_does_not_establish_prior_activity', + 'unknown_includes_wrapper_reasons', + 'no_provider_or_region', +] as const; + +const AGGREGATE_SERVICE = 'cloud-agent-next'; + +export type OutcomeGeneration = (typeof EXPECTED_GENERATIONS)[number]; +export type OutcomeRole = 'initial' | 'follow_up'; + +export type FailureStageCount = { stage: string; count: number }; +export type FailureStageCodeCount = { stage: string; code: string; count: number }; +export type SessionSetupFailureCount = { stage: string; code: string; count: number }; + +export type RoleOutcome = { + completed: number; + platformFailed: number; + userFailed: number; + unknownFailed: number; + interrupted: number; + settled: number; + allFailed: number; + platformFailureShare: number | null; + unknownClassificationShare: number | null; + failureStages: FailureStageCount[]; + failureStageCodes: FailureStageCodeCount[]; +}; + +export type GenerationOutcome = { + generation: OutcomeGeneration; + runRowsObserved: boolean; + initial: RoleOutcome; + followUp: RoleOutcome; + totals: RoleOutcome; + distinctPlatformAffectedSessions: number; + distinctUnknownAffectedSessions: number; + sessionSetupFailures: SessionSetupFailureCount[]; + sessionSetupFailureCount: number; +}; + +export type OutcomeWindow = { windowMinutes: number; start: string; end: string }; + +type RunCountRow = { + generation: OutcomeGeneration; + role: OutcomeRole; + status: CloudAgentSessionRunStatus; + responsibility: string; + failureStage: string; + failureCode: string; + runCount: number; +}; + +type DistinctAffectedRow = { + generation: OutcomeGeneration; + distinctPlatformAffectedSessions: number; + distinctUnknownAffectedSessions: number; +}; + +type SessionSetupFailureRow = { + generation: OutcomeGeneration; + failureStage: string; + failureCode: string; + failureCount: number; +}; + +type DatabaseTransaction = Parameters[0]>[0]; + +const unknownResponsibilityCondition: SQL = sql`(${cloud_agent_session_runs.failure_responsibility} is null or ${cloud_agent_session_runs.failure_responsibility} not in ('platform', 'user'))`; + +function generationExpression(sessionId: AnyColumn): SQL { + return sql`case when left(${sessionId}, ${CONTROL_PLANE_SESSION_PREFIX.length}) = ${CONTROL_PLANE_SESSION_PREFIX} then 'control' else 'legacy' end`; +} + +const roleExpression: SQL = sql`case when ${cloud_agent_session_runs.message_id} = ${cloud_agent_sessions.initial_message_id} then 'initial' else 'follow_up' end`; + +const responsibilityBucketExpression: SQL = sql`case when ${unknownResponsibilityCondition} then 'unknown' else ${cloud_agent_session_runs.failure_responsibility} end`; + +const runFailureStageExpression: SQL = sql`coalesce(${cloud_agent_session_runs.failure_stage}, 'unknown')`; +const runFailureCodeExpression: SQL = sql`coalesce(${cloud_agent_session_runs.failure_code}, 'unclassified')`; +const sessionFailureStageExpression: SQL = sql`coalesce(${cloud_agent_sessions.failure_stage}, 'unknown')`; +const sessionFailureCodeExpression: SQL = sql`coalesce(${cloud_agent_sessions.failure_code}, 'unclassified')`; + +function retainedWindow( + timeColumn: AnyColumn, + window: OutcomeWindow, + retentionCutoffIso: string +): SQL { + return ( + and( + gte(timeColumn, window.start), + lt(timeColumn, window.end), + gt(cloud_agent_sessions.created_at, retentionCutoffIso) + ) ?? sql`true` + ); +} + +async function readRunCounts( + tx: DatabaseTransaction, + window: OutcomeWindow, + retentionCutoffIso: string +): Promise { + const generation = generationExpression(cloud_agent_session_runs.cloud_agent_session_id); + return tx + .select({ + generation, + role: roleExpression, + status: cloud_agent_session_runs.status, + responsibility: responsibilityBucketExpression, + failureStage: runFailureStageExpression, + failureCode: runFailureCodeExpression, + runCount: sql`count(*)::int`, + }) + .from(cloud_agent_session_runs) + .innerJoin( + cloud_agent_sessions, + eq( + cloud_agent_sessions.cloud_agent_session_id, + cloud_agent_session_runs.cloud_agent_session_id + ) + ) + .where(retainedWindow(cloud_agent_session_runs.terminal_at, window, retentionCutoffIso)) + .groupBy(sql`1`, sql`2`, sql`3`, sql`4`, sql`5`, sql`6`); +} + +async function readDistinctAffectedSessions( + tx: DatabaseTransaction, + window: OutcomeWindow, + retentionCutoffIso: string +): Promise { + return tx + .select({ + generation: generationExpression(cloud_agent_session_runs.cloud_agent_session_id), + distinctPlatformAffectedSessions: sql`(count(distinct ${cloud_agent_session_runs.cloud_agent_session_id}) filter (where ${cloud_agent_session_runs.failure_responsibility} = 'platform'))::int`, + distinctUnknownAffectedSessions: sql`(count(distinct ${cloud_agent_session_runs.cloud_agent_session_id}) filter (where ${unknownResponsibilityCondition}))::int`, + }) + .from(cloud_agent_session_runs) + .innerJoin( + cloud_agent_sessions, + eq( + cloud_agent_sessions.cloud_agent_session_id, + cloud_agent_session_runs.cloud_agent_session_id + ) + ) + .where( + and( + eq(cloud_agent_session_runs.status, 'failed'), + retainedWindow(cloud_agent_session_runs.terminal_at, window, retentionCutoffIso) + ) + ) + .groupBy(sql`1`); +} + +async function readSessionSetupFailures( + tx: DatabaseTransaction, + window: OutcomeWindow, + retentionCutoffIso: string +): Promise { + const generation = generationExpression(cloud_agent_sessions.cloud_agent_session_id); + return tx + .select({ + generation, + failureStage: sessionFailureStageExpression, + failureCode: sessionFailureCodeExpression, + failureCount: sql`count(*)::int`, + }) + .from(cloud_agent_sessions) + .where(retainedWindow(cloud_agent_sessions.failure_at, window, retentionCutoffIso)) + .groupBy(sql`1`, sql`2`, sql`3`); +} + +function compareByStage(left: { stage: string }, right: { stage: string }): number { + return left.stage < right.stage ? -1 : left.stage > right.stage ? 1 : 0; +} + +function compareByStageCode( + left: { stage: string; code: string }, + right: { stage: string; code: string } +): number { + const byStage = compareByStage(left, right); + if (byStage !== 0) return byStage; + return left.code < right.code ? -1 : left.code > right.code ? 1 : 0; +} + +function roleOutcome(rows: RunCountRow[]): RoleOutcome { + let completed = 0; + let interrupted = 0; + let platformFailed = 0; + let userFailed = 0; + let unknownFailed = 0; + const stageTotals = new Map(); + const stageCodes = new Map(); + + for (const row of rows) { + if (row.status === 'completed') { + completed += row.runCount; + continue; + } + if (row.status === 'interrupted') { + interrupted += row.runCount; + continue; + } + if (row.status !== 'failed') continue; + + if (row.responsibility === 'platform') platformFailed += row.runCount; + else if (row.responsibility === 'user') userFailed += row.runCount; + else unknownFailed += row.runCount; + + stageTotals.set(row.failureStage, (stageTotals.get(row.failureStage) ?? 0) + row.runCount); + const key = `${row.failureStage}\u0000${row.failureCode}`; + const existing = stageCodes.get(key); + if (existing) existing.count += row.runCount; + else + stageCodes.set(key, { + stage: row.failureStage, + code: row.failureCode, + count: row.runCount, + }); + } + + const allFailed = platformFailed + userFailed + unknownFailed; + const settled = completed + allFailed; + + return { + completed, + platformFailed, + userFailed, + unknownFailed, + interrupted, + settled, + allFailed, + platformFailureShare: settled > 0 ? platformFailed / settled : null, + unknownClassificationShare: allFailed > 0 ? unknownFailed / allFailed : null, + failureStages: [...stageTotals] + .map(([stage, count]) => ({ stage, count })) + .sort(compareByStage), + failureStageCodes: [...stageCodes.values()].sort(compareByStageCode), + }; +} + +export function assembleGenerationAggregates( + runCounts: RunCountRow[], + distinctAffectedSessions: DistinctAffectedRow[], + sessionSetupFailures: SessionSetupFailureRow[] +): GenerationOutcome[] { + return EXPECTED_GENERATIONS.map(generation => { + const generationRuns = runCounts.filter(row => row.generation === generation); + const distinct = distinctAffectedSessions.find(row => row.generation === generation); + const setupFailures = sessionSetupFailures + .filter(row => row.generation === generation) + .map(row => ({ stage: row.failureStage, code: row.failureCode, count: row.failureCount })) + .sort(compareByStageCode); + return { + generation, + runRowsObserved: generationRuns.length > 0, + initial: roleOutcome(generationRuns.filter(row => row.role === 'initial')), + followUp: roleOutcome(generationRuns.filter(row => row.role === 'follow_up')), + totals: roleOutcome(generationRuns), + distinctPlatformAffectedSessions: distinct?.distinctPlatformAffectedSessions ?? 0, + distinctUnknownAffectedSessions: distinct?.distinctUnknownAffectedSessions ?? 0, + sessionSetupFailures: setupFailures, + sessionSetupFailureCount: setupFailures.reduce((sum, row) => sum + row.count, 0), + }; + }); +} + +export async function readOutcomeAggregate( + db: WorkerDb, + input: { window: OutcomeWindow; retentionCutoff: string } +): Promise { + return db.transaction( + async tx => { + const runCounts = await readRunCounts(tx, input.window, input.retentionCutoff); + const distinctAffectedSessions = await readDistinctAffectedSessions( + tx, + input.window, + input.retentionCutoff + ); + const sessionSetupFailures = await readSessionSetupFailures( + tx, + input.window, + input.retentionCutoff + ); + return assembleGenerationAggregates( + runCounts, + distinctAffectedSessions, + sessionSetupFailures + ); + }, + { isolationLevel: 'repeatable read', accessMode: 'read only' } + ); +} + +function evaluationEnvelope(input: { + window: OutcomeWindow; + collectionTimestamp: string; + queryElapsedMs: number; + retentionCutoff: string; +}): Record { + return { + metric: AGGREGATE_METRIC, + logTag: AGGREGATE_METRIC, + contractVersion: AGGREGATE_CONTRACT_VERSION, + service: AGGREGATE_SERVICE, + environment: null, + evaluationId: `${AGGREGATE_METRIC}:${input.window.windowMinutes}:${input.window.end}`, + collectionTimestamp: input.collectionTimestamp, + queryElapsedMs: input.queryElapsedMs, + windowStart: input.window.start, + windowEnd: input.window.end, + windowMinutes: input.window.windowMinutes, + reportingDelayMs: REPORTING_DELAY_ALLOWANCE_MS, + coverage: COVERAGE_LABEL, + retentionCutoff: input.retentionCutoff, + expectedGenerations: [...EXPECTED_GENERATIONS], + limitations: [...OUTCOME_AGGREGATE_LIMITATIONS], + }; +} + +export async function runCloudAgentOutcomeCollection(env: Env, now = new Date()): Promise { + const db = getPgDb(env); + const collectionTimestamp = now.toISOString(); + const retentionCutoffIso = retentionCutoff(collectionTimestamp); + const windowEnd = new Date(now.getTime() - REPORTING_DELAY_ALLOWANCE_MS).toISOString(); + + for (const windowMinutes of OUTCOME_WINDOW_MINUTES) { + const window: OutcomeWindow = { + windowMinutes, + start: new Date(Date.parse(windowEnd) - windowMinutes * 60_000).toISOString(), + end: windowEnd, + }; + const evaluatedAt = Date.now(); + try { + const generations = await readOutcomeAggregate(db, { + window, + retentionCutoff: retentionCutoffIso, + }); + logger + .withFields({ + ...evaluationEnvelope({ + window, + collectionTimestamp, + queryElapsedMs: Date.now() - evaluatedAt, + retentionCutoff: retentionCutoffIso, + }), + collectionStatus: 'complete', + generations, + }) + .info('Cloud Agent outcome aggregate'); + } catch { + logger + .withFields({ + ...evaluationEnvelope({ + window, + collectionTimestamp, + queryElapsedMs: Date.now() - evaluatedAt, + retentionCutoff: retentionCutoffIso, + }), + collectionStatus: 'failed', + failureKind: 'db_query_failed', + }) + .error('Cloud Agent outcome aggregate failed'); + return; + } + } +} diff --git a/services/cloud-agent-next/src/telemetry/report-store.ts b/services/cloud-agent-next/src/telemetry/report-store.ts index 01039f588a2..85164b6b89f 100644 --- a/services/cloud-agent-next/src/telemetry/report-store.ts +++ b/services/cloud-agent-next/src/telemetry/report-store.ts @@ -93,7 +93,7 @@ type StoredRunRow = { errorExpiresAt: string | null; }; -function retentionCutoff(now: string): string { +export function retentionCutoff(now: string): string { const cutoff = new Date(now); cutoff.setUTCDate(cutoff.getUTCDate() - CLOUD_AGENT_REPORT_RETENTION_DAYS); return cutoff.toISOString(); diff --git a/services/cloud-agent-next/test/postgres/outcome-aggregate.postgres.test.ts b/services/cloud-agent-next/test/postgres/outcome-aggregate.postgres.test.ts new file mode 100644 index 00000000000..1c62031ddac --- /dev/null +++ b/services/cloud-agent-next/test/postgres/outcome-aggregate.postgres.test.ts @@ -0,0 +1,727 @@ +import { randomUUID } from 'node:crypto'; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from 'vitest'; +import { createDrizzleClient, getWorkerDb, type WorkerDb } from '@kilocode/db/client'; +import { eq } from 'drizzle-orm'; +import { cloud_agent_session_runs, cloud_agent_sessions } from '@kilocode/db/schema'; +import type { CloudAgentRunStateReport } from '@kilocode/worker-utils/cloud-agent-queue-report'; +import { logger } from '../../src/logger.js'; +import { createCloudAgentReportStore } from '../../src/telemetry/report-store.js'; +import { + readOutcomeAggregate, + runCloudAgentOutcomeCollection, + type GenerationOutcome, + type OutcomeWindow, +} from '../../src/telemetry/outcome-aggregate.js'; +import { sessionPlaneFromId } from '../../src/session-plane.js'; + +const { getPgDbMock } = vi.hoisted(() => ({ getPgDbMock: vi.fn() })); + +vi.mock('../../src/db/pg.js', () => ({ getPgDb: getPgDbMock })); + +const AGGREGATE_RETENTION_CUTOFF = '2026-01-01T00:00:00.000Z'; +const WINDOW_BASE = Date.UTC(2026, 1, 1, 0, 0, 0, 0); + +let windowOffsetMinutes = 0; +let connectionString: string; +let reader: WorkerDb; +let writer: ReturnType; +let store: ReturnType; +let writerStore: ReturnType; +const trackedSessionIds: string[] = []; + +function requirePostgresUrl(): string { + const url = process.env.POSTGRES_URL; + if (!url) throw new Error('Set POSTGRES_URL to a migrated test database'); + return url; +} + +function nextWindow(): OutcomeWindow { + const start = new Date(WINDOW_BASE + windowOffsetMinutes * 60_000).toISOString(); + windowOffsetMinutes += 30; + return { windowMinutes: 10, start, end: new Date(Date.parse(start) + 10 * 60_000).toISOString() }; +} + +const uniqueKiloSessionId = () => `ses_${randomUUID().replace(/-/g, '').slice(0, 26)}`; +const uniqueMessageId = () => `msg_${randomUUID().replace(/-/g, '')}`; +const uniqueSessionId = (plane: 'agent' | 'workspace') => `${plane}_${randomUUID()}`; + +function report( + sessionId: string, + run: CloudAgentRunStateReport['run'], + anchor?: { kiloSessionId: string; initialMessageId: string; reportingCreatedAt: string } +): CloudAgentRunStateReport { + return { + version: 1, + type: 'run.state', + occurredAt: run.terminalAt ?? run.queuedAt ?? new Date().toISOString(), + session: { cloudAgentSessionId: sessionId, ...anchor }, + run, + }; +} + +function completedRun(messageId: string, terminalAt: string): CloudAgentRunStateReport['run'] { + return { messageId, status: 'completed', queuedAt: terminalAt, terminalAt }; +} + +function platformFailedRun(messageId: string, terminalAt: string): CloudAgentRunStateReport['run'] { + return { + messageId, + status: 'failed', + queuedAt: terminalAt, + terminalAt, + failureStage: 'pre_dispatch', + failureCode: 'sandbox_connect_failed', + failureResponsibility: 'platform', + failureReason: 'sandbox_connectivity', + }; +} + +function unclassifiedFailedRun( + messageId: string, + terminalAt: string +): CloudAgentRunStateReport['run'] { + return { + messageId, + status: 'failed', + queuedAt: terminalAt, + terminalAt, + failureStage: 'unknown', + failureCode: 'unclassified', + }; +} + +async function createSession( + sessionId: string, + occurredAt: string, + initialMessageId = uniqueMessageId() +): Promise { + trackedSessionIds.push(sessionId); + await store.createSessionReport({ + cloudAgentSessionId: sessionId, + kiloSessionId: uniqueKiloSessionId(), + initialMessageId, + occurredAt, + }); + return initialMessageId; +} + +async function saveReport( + sessionId: string, + run: CloudAgentRunStateReport['run'], + now: string, + anchor?: { kiloSessionId: string; initialMessageId: string; reportingCreatedAt: string } +): Promise { + trackedSessionIds.push(sessionId); + await store.saveReport(report(sessionId, run, anchor), now); +} + +async function insertSession(sessionId: string, createdAt: string): Promise { + trackedSessionIds.push(sessionId); + const initialMessageId = uniqueMessageId(); + await writer.db.insert(cloud_agent_sessions).values({ + cloud_agent_session_id: sessionId, + kilo_session_id: uniqueKiloSessionId(), + initial_message_id: initialMessageId, + created_at: createdAt, + }); + return initialMessageId; +} + +async function insertRun(values: { + cloudAgentSessionId: string; + messageId: string; + status: 'completed' | 'failed' | 'interrupted'; + terminalAt: string | null; + failureStage?: string; + failureCode?: string; + failureResponsibility?: string; + failureReason?: string; +}): Promise { + await writer.db.insert(cloud_agent_session_runs).values({ + cloud_agent_session_id: values.cloudAgentSessionId, + message_id: values.messageId, + status: values.status, + queued_at: values.terminalAt, + terminal_at: values.terminalAt, + failure_stage: values.failureStage ?? null, + failure_code: values.failureCode ?? null, + failure_responsibility: values.failureResponsibility ?? null, + failure_reason: values.failureReason ?? null, + }); +} + +async function aggregate( + window: OutcomeWindow, + retentionCutoff = AGGREGATE_RETENTION_CUTOFF +): Promise { + return readOutcomeAggregate(reader, { window, retentionCutoff }); +} + +function legacy(generations: GenerationOutcome[]): GenerationOutcome { + return generations[0]; +} + +function control(generations: GenerationOutcome[]): GenerationOutcome { + return generations[1]; +} + +function createBarrier() { + let signal!: () => void; + let release!: () => void; + const reached = new Promise(resolve => { + signal = resolve; + }); + const released = new Promise(resolve => { + release = resolve; + }); + return { reached, released, signal, release }; +} + +function pauseAfterQuery( + builder: object, + barrier: { signal: () => void; released: Promise } +): object { + return new Proxy(builder, { + get(target, property) { + if (property === 'then') { + return ( + onFulfilled: (value: unknown) => unknown, + onRejected: (reason: unknown) => unknown + ) => + (target as PromiseLike).then(async value => { + barrier.signal(); + await barrier.released; + return onFulfilled(value); + }, onRejected); + } + const value = Reflect.get(target, property, target); + if (typeof value !== 'function') return value; + return (...args: unknown[]) => + pauseAfterQuery((value as (...a: unknown[]) => object).apply(target, args), barrier); + }, + }); +} + +function withQ1Barrier( + db: WorkerDb, + barrier: { signal: () => void; released: Promise } +): WorkerDb { + return new Proxy(db, { + get(target, property) { + if (property !== 'transaction') return Reflect.get(target, property, target); + const transaction = Reflect.get(target, property, target) as ( + operation: (tx: unknown) => Promise, + options?: unknown + ) => Promise; + return (operation: (tx: unknown) => Promise, options?: unknown) => { + let firstQuery = true; + return transaction.call( + target, + (tx: unknown) => + operation( + new Proxy(tx as object, { + get(txTarget, txProperty) { + const txValue = Reflect.get(txTarget, txProperty, txTarget); + if (txProperty !== 'select' || typeof txValue !== 'function') return txValue; + return (...args: unknown[]) => { + const builder = (txValue as (...a: unknown[]) => object).apply(txTarget, args); + if (!firstQuery) return builder; + firstQuery = false; + return pauseAfterQuery(builder, barrier); + }; + }, + }) + ), + options + ); + }; + }, + }) as WorkerDb; +} + +function failAtQuery( + db: WorkerDb, + ordinal: number, + error: Error +): { db: WorkerDb; state: { selectCalls: number } } { + const state = { selectCalls: 0 }; + const instrumented = new Proxy(db, { + get(target, property) { + if (property !== 'transaction') return Reflect.get(target, property, target); + const transaction = Reflect.get(target, property, target) as ( + operation: (tx: unknown) => Promise, + options?: unknown + ) => Promise; + return (operation: (tx: unknown) => Promise, options?: unknown) => { + let selects = 0; + return transaction.call( + target, + (tx: unknown) => + operation( + new Proxy(tx as object, { + get(txTarget, txProperty) { + const txValue = Reflect.get(txTarget, txProperty, txTarget); + if (txProperty !== 'select' || typeof txValue !== 'function') return txValue; + return (...args: unknown[]) => { + selects += 1; + state.selectCalls += 1; + if (selects === ordinal) throw error; + return (txValue as (...a: unknown[]) => object).apply(txTarget, args); + }; + }, + }) + ), + options + ); + }; + }, + }) as WorkerDb; + return { db: instrumented, state }; +} + +beforeAll(() => { + connectionString = requirePostgresUrl(); + reader = getWorkerDb(connectionString); + writer = createDrizzleClient({ connectionString, ssl: false }); + store = createCloudAgentReportStore(reader); + writerStore = createCloudAgentReportStore(writer.db); +}); + +afterEach(async () => { + const ids = trackedSessionIds.splice(0); + if (ids.length === 0) return; + for (const id of ids) { + await writer.db + .delete(cloud_agent_sessions) + .where(eq(cloud_agent_sessions.cloud_agent_session_id, id)); + } +}); + +afterAll(async () => { + await writer.pool.end(); + const readerPool = (reader as unknown as { $client?: { end: () => Promise } }).$client; + await readerPool?.end(); +}); + +describe('cloud agent outcome aggregate against PostgreSQL', () => { + it('counts a run whose terminal_at equals the window start and excludes the window end', async () => { + const window = nextWindow(); + const startSession = uniqueSessionId('agent'); + await saveReport( + startSession, + completedRun(await createSession(startSession, window.start), window.start), + window.end + ); + + const endSession = uniqueSessionId('agent'); + await saveReport( + endSession, + completedRun(await createSession(endSession, window.start), window.end), + window.end + ); + + const result = await aggregate(window); + expect(legacy(result).totals.completed).toBe(1); + expect(legacy(result).runRowsObserved).toBe(true); + }); + + it('merges duplicate and out-of-order snapshots into one terminal turn', async () => { + const window = nextWindow(); + const sessionId = uniqueSessionId('agent'); + const initial = await createSession(sessionId, window.start); + + await saveReport( + sessionId, + { messageId: initial, status: 'queued', queuedAt: window.start }, + window.end + ); + await saveReport(sessionId, platformFailedRun(initial, window.start), window.end); + await saveReport( + sessionId, + { messageId: initial, status: 'queued', queuedAt: window.start }, + window.end + ); + await saveReport(sessionId, platformFailedRun(initial, window.start), window.end); + + const rows = await writer.db + .select({ messageId: cloud_agent_session_runs.message_id }) + .from(cloud_agent_session_runs) + .where(eq(cloud_agent_session_runs.cloud_agent_session_id, sessionId)); + expect(rows).toHaveLength(1); + + const result = await aggregate(window); + expect(legacy(result).totals.platformFailed).toBe(1); + expect(legacy(result).totals.allFailed).toBe(1); + expect(legacy(result).distinctPlatformAffectedSessions).toBe(1); + }); + + it('counts a run with null responsibility as unknown in both the run count and the distinct sessions', async () => { + const window = nextWindow(); + const sessionId = uniqueSessionId('agent'); + const initial = await createSession(sessionId, window.start); + await saveReport(sessionId, unclassifiedFailedRun(initial, window.start), window.end); + + const result = await aggregate(window); + expect(legacy(result).totals.unknownFailed).toBe(1); + expect(legacy(result).totals.allFailed).toBe(1); + expect(legacy(result).distinctUnknownAffectedSessions).toBe(1); + }); + + it('emits the unclassified failure-code sentinel for a failed run with no classification', async () => { + const window = nextWindow(); + const sessionId = uniqueSessionId('agent'); + const initial = await createSession(sessionId, window.start); + await saveReport( + sessionId, + { messageId: initial, status: 'failed', queuedAt: window.start, terminalAt: window.start }, + window.end + ); + + const result = await aggregate(window); + expect(legacy(result).totals.failureStages).toEqual([{ stage: 'unknown', count: 1 }]); + expect(legacy(result).totals.failureStageCodes).toEqual([ + { stage: 'unknown', code: 'unclassified', count: 1 }, + ]); + }); + + it('counts one distinct platform-affected session across many failing turns', async () => { + const window = nextWindow(); + const sessionId = uniqueSessionId('agent'); + const initial = await createSession(sessionId, window.start); + await saveReport(sessionId, platformFailedRun(initial, window.start), window.end); + await saveReport(sessionId, platformFailedRun(uniqueMessageId(), window.start), window.end); + await saveReport(sessionId, platformFailedRun(uniqueMessageId(), window.start), window.end); + + const result = await aggregate(window); + expect(legacy(result).totals.platformFailed).toBe(3); + expect(legacy(result).distinctPlatformAffectedSessions).toBe(1); + }); + + it('labels the run whose message_id matches initial_message_id as initial and others as follow-up', async () => { + const window = nextWindow(); + const sessionId = uniqueSessionId('agent'); + const initial = await createSession(sessionId, window.start); + await saveReport(sessionId, platformFailedRun(initial, window.start), window.end); + await saveReport(sessionId, platformFailedRun(uniqueMessageId(), window.start), window.end); + + const missingInitialSession = uniqueSessionId('agent'); + await createSession(missingInitialSession, window.start); + await saveReport( + missingInitialSession, + platformFailedRun(uniqueMessageId(), window.start), + window.end + ); + + const result = await aggregate(window); + expect(legacy(result).initial.platformFailed).toBe(1); + expect(legacy(result).followUp.platformFailed).toBe(2); + expect(legacy(result).totals.platformFailed).toBe(3); + }); + + it('classifies the first admitted run of an anchor-created control-plane parent as initial', async () => { + const window = nextWindow(); + const sessionId = uniqueSessionId('workspace'); + const initial = uniqueMessageId(); + await saveReport(sessionId, completedRun(initial, window.start), window.end, { + kiloSessionId: uniqueKiloSessionId(), + initialMessageId: initial, + reportingCreatedAt: window.start, + }); + + const result = await aggregate(window); + expect(control(result).totals.completed).toBe(1); + expect(control(result).initial.completed).toBe(1); + expect(control(result).runRowsObserved).toBe(true); + }); + + it('reports interrupted turns separately from settled and failed turns', async () => { + const window = nextWindow(); + const sessionId = uniqueSessionId('agent'); + const initial = await createSession(sessionId, window.start); + await saveReport( + sessionId, + { + messageId: initial, + status: 'interrupted', + queuedAt: window.start, + terminalAt: window.start, + failureStage: 'interruption', + failureCode: 'user_interrupt', + failureResponsibility: 'user', + failureReason: 'user_interrupt', + }, + window.end + ); + await saveReport(sessionId, completedRun(uniqueMessageId(), window.start), window.end); + + const result = await aggregate(window); + expect(legacy(result).totals.completed).toBe(1); + expect(legacy(result).totals.interrupted).toBe(1); + expect(legacy(result).totals.settled).toBe(1); + expect(legacy(result).totals.allFailed).toBe(0); + }); + + it('counts an unexpected responsibility text value as unknown', async () => { + const window = nextWindow(); + const sessionId = uniqueSessionId('agent'); + const initial = await insertSession(sessionId, window.start); + await insertRun({ + cloudAgentSessionId: sessionId, + messageId: initial, + status: 'failed', + terminalAt: window.start, + failureStage: 'agent_activity', + failureCode: 'assistant_error', + failureResponsibility: 'provider', + failureReason: 'assistant_unknown', + }); + + const result = await aggregate(window); + expect(legacy(result).totals.unknownFailed).toBe(1); + expect(legacy(result).totals.allFailed).toBe(1); + expect(legacy(result).distinctUnknownAffectedSessions).toBe(1); + }); + + it('retains only sessions created strictly after the retention cutoff', async () => { + const window = nextWindow(); + const cutoff = '2026-01-15T00:00:00.000Z'; + const atCutoff = uniqueSessionId('agent'); + const older = uniqueSessionId('agent'); + const newer = uniqueSessionId('agent'); + + await insertSession(atCutoff, cutoff); + await insertSession(older, new Date(Date.parse(cutoff) - 1000).toISOString()); + await insertSession(newer, new Date(Date.parse(cutoff) + 1000).toISOString()); + await insertRun({ + cloudAgentSessionId: atCutoff, + messageId: uniqueMessageId(), + status: 'completed', + terminalAt: window.start, + }); + await insertRun({ + cloudAgentSessionId: older, + messageId: uniqueMessageId(), + status: 'completed', + terminalAt: window.start, + }); + await insertRun({ + cloudAgentSessionId: newer, + messageId: uniqueMessageId(), + status: 'completed', + terminalAt: window.start, + }); + + const result = await aggregate(window, cutoff); + expect(legacy(result).totals.completed).toBe(1); + }); + + it('splits generations by the workspace_ prefix in agreement with sessionPlaneFromId', async () => { + const window = nextWindow(); + const sessionIds = [ + uniqueSessionId('workspace'), + uniqueSessionId('agent'), + 'workspace', + `Workspace_${randomUUID()}`, + 'agent_', + '', + ]; + + for (const sessionId of sessionIds) { + const initial = await insertSession(sessionId, window.start); + await insertRun({ + cloudAgentSessionId: sessionId, + messageId: initial, + status: 'completed', + terminalAt: window.start, + }); + } + + const result = await aggregate(window); + const expectedLegacy = sessionIds.filter(id => sessionPlaneFromId(id) === 'legacy').length; + const expectedControl = sessionIds.filter(id => sessionPlaneFromId(id) === 'control').length; + expect(expectedControl).toBe(1); + expect(legacy(result).totals.completed).toBe(expectedLegacy); + expect(control(result).totals.completed).toBe(expectedControl); + }); + + it('returns null shares for a window with no settled turns', async () => { + const window = nextWindow(); + const result = await aggregate(window); + + expect(result.map(entry => entry.generation)).toEqual(['legacy', 'control']); + expect(legacy(result).runRowsObserved).toBe(false); + expect(control(result).runRowsObserved).toBe(false); + expect(legacy(result).totals.settled).toBe(0); + expect(legacy(result).totals.platformFailureShare).toBeNull(); + expect(legacy(result).totals.unknownClassificationShare).toBeNull(); + expect(control(result).totals.platformFailureShare).toBeNull(); + expect(control(result).totals.unknownClassificationShare).toBeNull(); + }); + + it('reports session setup failures separately from run outcomes', async () => { + const window = nextWindow(); + const sessionId = uniqueSessionId('agent'); + const initial = await createSession(sessionId, window.start); + await store.recordSessionFailure({ + cloudAgentSessionId: sessionId, + occurredAt: window.start, + failure: { stage: 'registration', code: 'do_registration_rejected' }, + }); + await saveReport(sessionId, completedRun(initial, window.start), window.end); + + const result = await aggregate(window); + expect(legacy(result).sessionSetupFailures).toEqual([ + { stage: 'registration', code: 'do_registration_rejected', count: 1 }, + ]); + expect(legacy(result).sessionSetupFailureCount).toBe(1); + expect(legacy(result).totals.completed).toBe(1); + expect(legacy(result).totals.settled).toBe(1); + }); + + it('counts a pre-dispatch platform failure without a dispatch acceptance fact', async () => { + const window = nextWindow(); + const sessionId = uniqueSessionId('agent'); + const initial = await createSession(sessionId, window.start); + await saveReport(sessionId, platformFailedRun(initial, window.start), window.end); + + const result = await aggregate(window); + expect(legacy(result).totals.platformFailed).toBe(1); + expect(legacy(result).totals.settled).toBe(1); + expect(legacy(result).distinctPlatformAffectedSessions).toBe(1); + }); + + it('excludes queued and failed rows with no terminal_at', async () => { + const window = nextWindow(); + const sessionId = uniqueSessionId('agent'); + const initial = await insertSession(sessionId, window.start); + await insertRun({ + cloudAgentSessionId: sessionId, + messageId: initial, + status: 'completed', + terminalAt: null, + }); + await insertRun({ + cloudAgentSessionId: sessionId, + messageId: uniqueMessageId(), + status: 'failed', + terminalAt: null, + failureStage: 'pre_dispatch', + failureCode: 'sandbox_connect_failed', + failureResponsibility: 'platform', + failureReason: 'sandbox_connectivity', + }); + + const result = await aggregate(window); + expect(legacy(result).runRowsObserved).toBe(false); + expect(legacy(result).totals.completed).toBe(0); + expect(legacy(result).totals.platformFailed).toBe(0); + }); + + it('serializes real query results to JSON without a BigInt error', async () => { + const window = nextWindow(); + const sessionId = uniqueSessionId('agent'); + const initial = await createSession(sessionId, window.start); + await saveReport(sessionId, platformFailedRun(initial, window.start), window.end); + + const result = await aggregate(window); + expect(typeof legacy(result).totals.platformFailed).toBe('number'); + expect(() => JSON.stringify(result)).not.toThrow(); + expect(JSON.parse(JSON.stringify(result))[0].totals.platformFailed).toBe(1); + }); + + it('excludes a run failure and a setup failure committed after Q1 in one snapshot', async () => { + const window = nextWindow(); + const sessionId = uniqueSessionId('agent'); + const initial = await createSession(sessionId, window.start); + + const barrier = createBarrier(); + const inflight = readOutcomeAggregate(withQ1Barrier(reader, barrier), { + window, + retentionCutoff: AGGREGATE_RETENTION_CUTOFF, + }); + try { + await barrier.reached; + await writerStore.saveReport( + report(sessionId, platformFailedRun(initial, window.start)), + window.end + ); + await writerStore.recordSessionFailure({ + cloudAgentSessionId: sessionId, + occurredAt: window.start, + failure: { stage: 'registration', code: 'do_registration_rejected' }, + }); + + // Both commits are visible to a separate client before Q2/Q3 are released, + // so a READ COMMITTED transaction would observe them in Q2 (distinct + // platform sessions) and Q3 (session setup failures). Only the snapshot + // taken at Q1 can keep the in-flight evaluation at zero. + const committedRuns = await writer.db + .select({ messageId: cloud_agent_session_runs.message_id }) + .from(cloud_agent_session_runs) + .where(eq(cloud_agent_session_runs.cloud_agent_session_id, sessionId)); + expect(committedRuns).toHaveLength(1); + const [committedSession] = await writer.db + .select({ failureAt: cloud_agent_sessions.failure_at }) + .from(cloud_agent_sessions) + .where(eq(cloud_agent_sessions.cloud_agent_session_id, sessionId)); + expect(Date.parse(committedSession.failureAt ?? '')).toBe(Date.parse(window.start)); + + barrier.release(); + const generations = await inflight; + expect(legacy(generations).totals.platformFailed).toBe(0); + expect(legacy(generations).distinctPlatformAffectedSessions).toBe(0); + expect(legacy(generations).sessionSetupFailureCount).toBe(0); + } finally { + barrier.release(); + } + + const after = await aggregate(window); + expect(legacy(after).totals.platformFailed).toBe(1); + expect(legacy(after).distinctPlatformAffectedSessions).toBe(1); + expect(legacy(after).sessionSetupFailures).toEqual([ + { stage: 'registration', code: 'do_registration_rejected', count: 1 }, + ]); + expect(legacy(after).sessionSetupFailureCount).toBe(1); + }); + + it('rejects and emits one failed record when a later query fails after an earlier one succeeded', async () => { + const window = nextWindow(); + const injected = new Error('injected outcome Q3 failure'); + const instrumented = failAtQuery(reader, 3, injected); + + await expect( + readOutcomeAggregate(instrumented.db, { + window, + retentionCutoff: AGGREGATE_RETENTION_CUTOFF, + }) + ).rejects.toThrow('injected outcome Q3 failure'); + expect(instrumented.state.selectCalls).toBe(3); + + // The same max:1 connection answers the next query, so the failed + // transaction released the connection rather than leaving it unusable. + const healthy = await aggregate(window); + expect(legacy(healthy).runRowsObserved).toBe(false); + + const errorMock = vi.fn(); + const infoMock = vi.fn(); + const withFieldsSpy = vi + .spyOn(logger, 'withFields') + .mockReturnValue({ info: infoMock, error: errorMock } as never); + getPgDbMock.mockReturnValue(instrumented.db); + let records: Record[] = []; + try { + await runCloudAgentOutcomeCollection({} as never, new Date('2026-02-01T00:10:00.000Z')); + records = withFieldsSpy.mock.calls.map(([fields]) => fields as Record); + } finally { + getPgDbMock.mockReset(); + withFieldsSpy.mockRestore(); + } + + expect(records).toHaveLength(1); + expect(records[0].collectionStatus).toBe('failed'); + expect(records[0].failureKind).toBe('db_query_failed'); + expect(records[0].generations).toBeUndefined(); + expect(errorMock).toHaveBeenCalledTimes(1); + expect(infoMock).not.toHaveBeenCalled(); + }); +}); diff --git a/services/cloud-agent-next/test/postgres/setup.ts b/services/cloud-agent-next/test/postgres/setup.ts new file mode 100644 index 00000000000..cb57c96d322 --- /dev/null +++ b/services/cloud-agent-next/test/postgres/setup.ts @@ -0,0 +1,3 @@ +if (!process.env.POSTGRES_URL) { + throw new Error('Set POSTGRES_URL to a migrated test database before running the Postgres suite'); +} diff --git a/services/cloud-agent-next/vitest.postgres.config.ts b/services/cloud-agent-next/vitest.postgres.config.ts new file mode 100644 index 00000000000..3e9b28e7934 --- /dev/null +++ b/services/cloud-agent-next/vitest.postgres.config.ts @@ -0,0 +1,12 @@ +import { defineConfig } from 'vitest/config'; + +export default defineConfig({ + test: { + name: 'postgres', + globals: true, + environment: 'node', + include: ['test/postgres/**/*.test.ts'], + setupFiles: ['test/postgres/setup.ts'], + fileParallelism: false, + }, +}); diff --git a/services/cloud-agent-next/wrangler.jsonc b/services/cloud-agent-next/wrangler.jsonc index 0a0b0289727..d30d5bf249f 100644 --- a/services/cloud-agent-next/wrangler.jsonc +++ b/services/cloud-agent-next/wrangler.jsonc @@ -34,7 +34,7 @@ }, "logpush": true, "triggers": { - "crons": ["17 2 * * *"], + "crons": ["17 2 * * *", "*/3 * * * *"], }, "routes": [ { @@ -421,7 +421,7 @@ "dev": { "name": "cloud-agent-next-dev", "triggers": { - "crons": ["17 2 * * *"], + "crons": ["17 2 * * *", "*/3 * * * *"], }, "hyperdrive": [ {