feat(cloud-agent-next): collect deduplicated outcome aggregates

.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 != '[]' }}

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",

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);
   });
 });
 

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<Response> {
     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<void> {
-    await removeExpiredCloudAgentReportData(env);
+  async scheduled(controller: ScheduledController, env: Env): Promise<void> {
+    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,
+    });
   },
 };
 

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 = {