feat(cloud-agent-next): emit point-in-time Cloud Agent open stock metric

services/cloud-agent-next/src/server.test.ts

@@ -22,6 +22,7 @@ const {
   consumeCloudAgentReportBatchMock,
   removeExpiredCloudAgentReportDataMock,
   runCloudAgentOutcomeCollectionMock,
+  runCloudAgentOpenStockCollectionMock,
   requireCurrentSessionAccessMock,
   getPgDbMock,
   loggerWarnMock,
@@ -30,6 +31,7 @@ const {
   consumeCloudAgentReportBatchMock: vi.fn().mockResolvedValue(undefined),
   removeExpiredCloudAgentReportDataMock: vi.fn().mockResolvedValue(undefined),
   runCloudAgentOutcomeCollectionMock: vi.fn().mockResolvedValue(undefined),
+  runCloudAgentOpenStockCollectionMock: vi.fn().mockResolvedValue(undefined),
   requireCurrentSessionAccessMock: vi.fn(),
   getPgDbMock: vi.fn(),
   loggerWarnMock: vi.fn(),
@@ -92,6 +94,10 @@ vi.mock('./telemetry/outcome-aggregate.js', () => ({
   runCloudAgentOutcomeCollection: runCloudAgentOutcomeCollectionMock,
 }));
 
+vi.mock('./telemetry/open-stock.js', () => ({
+  runCloudAgentOpenStockCollection: runCloudAgentOpenStockCollectionMock,
+}));
+
 vi.mock('./middleware/auth.js', () => ({
   authMiddleware: vi.fn(),
 }));
@@ -282,6 +288,7 @@ beforeEach(() => {
   consumeCloudAgentReportBatchMock.mockClear();
   removeExpiredCloudAgentReportDataMock.mockClear();
   runCloudAgentOutcomeCollectionMock.mockClear();
+  runCloudAgentOpenStockCollectionMock.mockClear();
   loggerWarnMock.mockClear();
   getPgDbMock.mockReset();
   requireCurrentSessionAccessMock.mockReset().mockResolvedValue({
@@ -394,9 +401,10 @@ describe('server background reporting', () => {
     expect(removeExpiredCloudAgentReportDataMock).toHaveBeenCalledTimes(1);
     expect(removeExpiredCloudAgentReportDataMock).toHaveBeenCalledWith(env);
     expect(runCloudAgentOutcomeCollectionMock).not.toHaveBeenCalled();
+    expect(runCloudAgentOpenStockCollectionMock).not.toHaveBeenCalled();
   });
 
-  it('runs only the outcome collection on the 3-minute cron', async () => {
+  it('runs the outcome and open-stock collections on the 3-minute cron', async () => {
     const env = createEnv();
 
     await worker.scheduled(
@@ -406,16 +414,35 @@ describe('server background reporting', () => {
 
     expect(runCloudAgentOutcomeCollectionMock).toHaveBeenCalledTimes(1);
     expect(runCloudAgentOutcomeCollectionMock).toHaveBeenCalledWith(env);
+    expect(runCloudAgentOpenStockCollectionMock).toHaveBeenCalledTimes(1);
+    expect(runCloudAgentOpenStockCollectionMock).toHaveBeenCalledWith(env);
     expect(removeExpiredCloudAgentReportDataMock).not.toHaveBeenCalled();
   });
 
+  it('still runs the open-stock collection and preserves the outcome error when outcome collection rejects', async () => {
+    const env = createEnv();
+    const outcomeError = new Error('outcome collection failed');
+    runCloudAgentOutcomeCollectionMock.mockRejectedValueOnce(outcomeError);
+
+    await expect(
+      worker.scheduled(
+        { cron: OUTCOME_AGGREGATE_CRON } as ScheduledController,
+        env as unknown as Env
+      )
+    ).rejects.toBe(outcomeError);
+
+    expect(runCloudAgentOpenStockCollectionMock).toHaveBeenCalledTimes(1);
+    expect(runCloudAgentOpenStockCollectionMock).toHaveBeenCalledWith(env);
+  });
+
   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(runCloudAgentOpenStockCollectionMock).not.toHaveBeenCalled();
     expect(loggerWarnMock).toHaveBeenCalledWith(
       'Cloud Agent scheduled handler received an unrecognized cron',
       { cron: '0 0 * * 0' }

services/cloud-agent-next/src/server.ts

@@ -24,6 +24,7 @@ import {
   removeExpiredCloudAgentReportData,
 } from './telemetry/report-consumer.js';
 import { runCloudAgentOutcomeCollection } from './telemetry/outcome-aggregate.js';
+import { runCloudAgentOpenStockCollection } from './telemetry/open-stock.js';
 import { authMiddleware } from './middleware/auth.js';
 import { balanceMiddleware } from './middleware/balance.js';
 import { resolveTerminalWrapperClient } from './terminal/access.js';
@@ -1122,7 +1123,11 @@ export default {
       return;
     }
     if (controller.cron === OUTCOME_AGGREGATE_CRON) {
-      await runCloudAgentOutcomeCollection(env);
+      try {
+        await runCloudAgentOutcomeCollection(env);
+      } finally {
+        await runCloudAgentOpenStockCollection(env);
+      }
       return;
     }
     logger.warn('Cloud Agent scheduled handler received an unrecognized cron', {

services/cloud-agent-next/src/telemetry/open-stock.test.ts

@@ -0,0 +1,291 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+import { eq, sql, type SQL } 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 { COVERAGE_LABEL } from './outcome-aggregate.js';
+import {
+  OPEN_STOCK_CONTRACT_VERSION,
+  OPEN_STOCK_LIMITATIONS,
+  OPEN_STOCK_METRIC,
+  assembleOpenStock,
+  readOpenStock,
+  runCloudAgentOpenStockCollection,
+  type OpenStockQueryRow,
+} from './open-stock.js';
+
+const retentionCutoffIso = '2025-11-03T00:10:00.000Z';
+const now = '2026-02-01T00:10:00.000Z';
+
+type QueryResult = OpenStockQueryRow[] | Error;
+
+function queryRow(
+  input: Partial<OpenStockQueryRow> & {
+    generation: OpenStockQueryRow['generation'];
+    status: OpenStockQueryRow['status'];
+  }
+): OpenStockQueryRow {
+  return {
+    turns: 0,
+    sessions: 0,
+    oldestQueuedEpochMs: null,
+    oldestAcceptedEpochMs: null,
+    queuedMissingAgeTurns: 0,
+    acceptedMissingAgeTurns: 0,
+    ...input,
+  };
+}
+
+function makeDb(result: QueryResult) {
+  const state: {
+    fields?: Record<string, unknown>;
+    where?: unknown;
+    groupBy?: unknown[];
+  } = {};
+
+  const chain = {
+    from: () => chain,
+    innerJoin: () => chain,
+    where: (arg: unknown) => {
+      state.where = arg;
+      return chain;
+    },
+    groupBy: (...args: unknown[]) => {
+      state.groupBy = args;
+      return chain;
+    },
+    then(resolve: (value: unknown) => unknown, reject: (reason: unknown) => unknown): unknown {
+      if (result instanceof Error) return reject(result);
+      return resolve(result);
+    },
+  };
+
+  const db = {
+    select: (fields: Record<string, unknown>) => {
+      state.fields = fields;
+      return chain;
+    },
+  };
+
+  return { db, state };
+}
+
+const usageDb = () => getWorkerDb('postgres://unused:unused@localhost:0/unused');
+
+function renderSelect(fields: Record<string, unknown>) {
+  return usageDb()
+    .select(fields as never)
+    .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();
+}
+
+function renderWhere(where: unknown, groupBy: unknown[]) {
+  return usageDb()
+    .select({ marker: sql`1` })
+    .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(where as SQL)
+    .groupBy(...(groupBy as SQL[]))
+    .toSQL();
+}
+
+beforeEach(() => {
+  vi.clearAllMocks();
+  loggerMock.withFields.mockReturnValue(loggerMock);
+});
+
+describe('cloud agent open stock assembly', () => {
+  it('reports both generations as empty stock with null ages', () => {
+    const generations = assembleOpenStock([], now);
+
+    expect(generations.map(entry => entry.generation)).toEqual(['legacy', 'control']);
+    for (const entry of generations) {
+      expect(entry).toEqual({
+        generation: entry.generation,
+        queuedTurns: 0,
+        queuedSessions: 0,
+        acceptedTurns: 0,
+        acceptedSessions: 0,
+        oldestQueuedAgeMs: null,
+        oldestAcceptedAgeMs: null,
+        queuedMissingAgeTurns: 0,
+        acceptedMissingAgeTurns: 0,
+      });
+    }
+  });
+
+  it('preserves queued counts and the missing-age count without inventing an age', () => {
+    const generations = assembleOpenStock(
+      [
+        queryRow({
+          generation: 'legacy',
+          status: 'queued',
+          turns: 3,
+          sessions: 2,
+          oldestQueuedEpochMs: null,
+          queuedMissingAgeTurns: 3,
+        }),
+      ],
+      now
+    );
+
+    const legacy = generations[0];
+    expect(legacy.queuedTurns).toBe(3);
+    expect(legacy.queuedSessions).toBe(2);
+    expect(legacy.oldestQueuedAgeMs).toBeNull();
+    expect(legacy.queuedMissingAgeTurns).toBe(3);
+    expect(legacy.acceptedTurns).toBe(0);
+    expect(legacy.oldestAcceptedAgeMs).toBeNull();
+  });
+
+  it('derives queued and accepted ages from their own timestamps', () => {
+    const queuedEpochMs = Date.parse('2026-02-01T00:00:00.000Z');
+    const acceptedEpochMs = Date.parse('2026-01-31T23:55:00.000Z');
+    const generations = assembleOpenStock(
+      [
+        queryRow({
+          generation: 'legacy',
+          status: 'queued',
+          turns: 1,
+          sessions: 1,
+          oldestQueuedEpochMs: queuedEpochMs,
+        }),
+        queryRow({
+          generation: 'legacy',
+          status: 'accepted',
+          turns: 2,
+          sessions: 2,
+          oldestAcceptedEpochMs: acceptedEpochMs,
+        }),
+      ],
+      now
+    );
+
+    const legacy = generations[0];
+    expect(legacy.oldestQueuedAgeMs).toBe(Date.parse(now) - queuedEpochMs);
+    expect(legacy.oldestAcceptedAgeMs).toBe(Date.parse(now) - acceptedEpochMs);
+    expect(legacy.oldestAcceptedAgeMs).not.toBe(legacy.oldestQueuedAgeMs);
+  });
+});
+
+describe('cloud agent open stock query shape', () => {
+  it('narrows to non-terminal queued/accepted runs of retained sessions', async () => {
+    const fake = makeDb([]);
+    await readOpenStock(fake.db as never, { retentionCutoff: retentionCutoffIso });
+
+    const fields = fake.state.fields as Record<string, unknown>;
+    const selectSql = renderSelect(fields);
+    const whereSql = renderWhere(fake.state.where, fake.state.groupBy ?? []);
+
+    expect(selectSql.params).toEqual([10, 'workspace_']);
+    expect(selectSql.sql).toMatch(
+      /left\("cloud_agent_session_runs"\."cloud_agent_session_id", \$1\) = \$2/
+    );
+    expect(selectSql.sql).toContain(
+      'count(distinct "cloud_agent_session_runs"."cloud_agent_session_id"))::int'
+    );
+    expect(selectSql.sql).toContain('extract(epoch from');
+    expect(selectSql.sql).toContain('::double precision');
+    expect(selectSql.sql).toContain(
+      `count(*) filter (where "cloud_agent_session_runs"."status" = 'queued' and "cloud_agent_session_runs"."queued_at" is null)`
+    );
+    expect(selectSql.sql).toContain(
+      `count(*) filter (where "cloud_agent_session_runs"."status" = 'accepted' and "cloud_agent_session_runs"."dispatch_accepted_at" is null)`
+    );
+
+    expect(whereSql.sql).toContain('"status" in ($1, $2)');
+    expect(whereSql.sql).toContain('"terminal_at" is null');
+    expect(whereSql.sql).toContain('created_at" > $3');
+    expect(whereSql.sql).toContain('group by 1, 2');
+    expect(whereSql.params).toEqual(['queued', 'accepted', retentionCutoffIso]);
+  });
+});
+
+describe('runCloudAgentOpenStockCollection emission', () => {
+  it('emits one complete record with the stock envelope and two generations', async () => {
+    getPgDbMock.mockReturnValue(makeDb([]).db);
+
+    await runCloudAgentOpenStockCollection({} as never, new Date(now));
+
+    expect(loggerMock.info).toHaveBeenCalledTimes(1);
+    expect(loggerMock.error).not.toHaveBeenCalled();
+    const record = loggerMock.withFields.mock.calls[0]?.[0] as Record<string, unknown>;
+    expect(record.metric).toBe(OPEN_STOCK_METRIC);
+    expect(record.logTag).toBe(OPEN_STOCK_METRIC);
+    expect(record.contractVersion).toBe(OPEN_STOCK_CONTRACT_VERSION);
+    expect(record.service).toBe('cloud-agent-next');
+    expect(record.environment).toBeNull();
+    expect(record.evaluationId).toBe(`${OPEN_STOCK_METRIC}:${now}`);
+    expect(record.collectionTimestamp).toBe(now);
+    expect(typeof record.queryElapsedMs).toBe('number');
+    expect(record.coverage).toBe(COVERAGE_LABEL);
+    expect(record.retentionCutoff).toBe(retentionCutoffIso);
+    expect(record.expectedGenerations).toEqual(['legacy', 'control']);
+    expect(record.limitations).toEqual([...OPEN_STOCK_LIMITATIONS]);
+    expect(record.collectionStatus).toBe('complete');
+    expect(record.generations).toHaveLength(2);
+    expect(record.windowMinutes).toBeUndefined();
+    expect(record.windowStart).toBeUndefined();
+    expect(record.windowEnd).toBeUndefined();
+    expect(record.reportingDelayMs).toBeUndefined();
+  });
+
+  it('emits exactly one failed record with no generations when the query rejects', async () => {
+    getPgDbMock.mockReturnValue(makeDb(new Error('stock query failed')).db);
+
+    await expect(
+      runCloudAgentOpenStockCollection({} as never, new Date(now))
+    ).resolves.toBeUndefined();
+
+    expect(loggerMock.error).toHaveBeenCalledTimes(1);
+    expect(loggerMock.info).not.toHaveBeenCalled();
+    const record = loggerMock.withFields.mock.calls.at(-1)?.[0] as Record<string, unknown>;
+    expect(record.collectionStatus).toBe('failed');
+    expect(record.failureKind).toBe('db_query_failed');
+    expect(record.generations).toBeUndefined();
+    expect(record.limitations).toEqual([...OPEN_STOCK_LIMITATIONS]);
+  });
+
+  it('emits a failed record when the database binding throws', async () => {
+    getPgDbMock.mockImplementation(() => {
+      throw new Error('HYPERDRIVE binding missing');
+    });
+
+    await expect(
+      runCloudAgentOpenStockCollection({} as never, new Date(now))
+    ).resolves.toBeUndefined();
+
+    expect(loggerMock.error).toHaveBeenCalledTimes(1);
+    expect(loggerMock.info).not.toHaveBeenCalled();
+    const record = loggerMock.withFields.mock.calls.at(-1)?.[0] as Record<string, unknown>;
+    expect(record.collectionStatus).toBe('failed');
+    expect(record.failureKind).toBe('db_query_failed');
+    expect(record.generations).toBeUndefined();
+  });
+});