feat(billing): record Cloud Agent container usage

dev/local/services.ts

@@ -100,6 +100,7 @@ const serviceMeta: Record<string, ServiceMeta> = {
       'nextjs',
       'cloudflare-session-ingest',
       'cloudflare-git-token-service',
+      'container-usage-meter',
       'notifications',
     ],
     dir: 'services/cloud-agent-next',

packages/container-usage/src/context.ts

@@ -10,6 +10,7 @@ export const billingContextSchema = usageContextSchema
     measurementStarted: z.boolean(),
     nextSeq: z.number().int().positive().max(2_147_483_647).default(1),
     usageMeasuredAtMs: z.number().int().nonnegative().finite(),
+    stoppedObservedAtMs: z.number().int().nonnegative().finite().optional(),
     pendingHeartbeat: z
       .object({
         seq: z.number().int().positive().finite(),
@@ -69,6 +70,7 @@ export async function setBillingContext(
     measurementStarted: false,
     nextSeq: 1,
     usageMeasuredAtMs: Date.now(),
+    stoppedObservedAtMs: undefined,
     pendingHeartbeat: undefined,
     pendingStop: undefined,
   });

packages/container-usage/src/heartbeat.test.ts

@@ -107,6 +107,24 @@ describe('installBillingHeartbeat', () => {
     expect(Object.hasOwn(container, BILLING_HEARTBEAT_CALLBACK)).toBe(true);
   });
 
+  it('does not mark measurement started when initial scheduling fails', async () => {
+    const storage = memoryStorage();
+    await storedContext(storage);
+    const controller = installBillingHeartbeat(
+      {
+        deleteSchedules: vi.fn(),
+        getState: vi.fn(),
+        schedule: vi.fn(async () => {
+          throw new Error('schedule unavailable');
+        }) as Container['schedule'],
+      },
+      { client: usageClient('continue'), storage, enforceBudgetStop: vi.fn() }
+    );
+
+    await expect(controller.scheduleHeartbeat()).rejects.toThrow('schedule unavailable');
+    expect((await getBillingContext(storage))?.measurementStarted).toBe(false);
+  });
+
   it('keeps a stopped-state probe scheduled until stop is durably acknowledged', async () => {
     const storage = memoryStorage();
     await storedContext(storage);
@@ -129,6 +147,206 @@ describe('installBillingHeartbeat', () => {
     expect(schedule).toHaveBeenCalledWith(300, BILLING_HEARTBEAT_CALLBACK, expect.any(String));
   });
 
+  it('defers stopped-state closure when the producer owns an authoritative stop hook', async () => {
+    const storage = memoryStorage();
+    await storedContext(storage);
+    const schedule = vi.fn();
+    const recordStop = vi.fn<ContainerUsageRpcMethods['recordStop']>(async input => ({
+      intervalId: `${input.instanceId}:${input.startEpochMs}`,
+      durable: 'pg',
+      dedup: false,
+    }));
+    const client = new ContainerUsageClient(
+      {
+        recordStart: async () => ({
+          success: true,
+          ack: { intervalId: 'instance-1:123', durable: 'pg', dedup: false },
+        }),
+        recordHeartbeat: async () => ({
+          intervalId: 'instance-1:123',
+          durable: 'pg',
+          dedup: false,
+          budget: { verdict: 'continue' },
+        }),
+        recordStop,
+      },
+      { service: 'cloud-agent-next' }
+    );
+    const controller = installBillingHeartbeat(
+      {
+        deleteSchedules: vi.fn(),
+        getState: vi.fn(async () => ({
+          status: 'stopped_with_code' as const,
+          lastChange: Date.now(),
+          exitCode: 17,
+        })),
+        schedule: schedule as Container['schedule'],
+      },
+      { client, storage, stopOnStoppedState: false, enforceBudgetStop: vi.fn() }
+    );
+
+    await controller.billingHeartbeatTick();
+
+    expect(recordStop).not.toHaveBeenCalled();
+    expect(await getBillingContext(storage)).toBeDefined();
+    expect(schedule).toHaveBeenCalledWith(300, BILLING_HEARTBEAT_CALLBACK, expect.any(String));
+  });
+
+  it('closes a repeatedly stopped generation after the deferred stale grace', async () => {
+    const now = vi.spyOn(Date, 'now').mockReturnValue(1_000);
+    try {
+      const storage = memoryStorage();
+      await storedContext(storage);
+      const recordStop = vi.fn<ContainerUsageRpcMethods['recordStop']>(async input => ({
+        intervalId: `${input.instanceId}:${input.startEpochMs}`,
+        durable: 'pg',
+        dedup: false,
+      }));
+      const client = new ContainerUsageClient(
+        {
+          recordStart: async () => ({
+            success: true,
+            ack: { intervalId: 'instance-1:123', durable: 'pg', dedup: false },
+          }),
+          recordHeartbeat: async () => ({
+            intervalId: 'instance-1:123',
+            durable: 'pg',
+            dedup: false,
+            budget: { verdict: 'continue' },
+          }),
+          recordStop,
+        },
+        { service: 'cloud-agent-next' }
+      );
+      const controller = installBillingHeartbeat(
+        {
+          deleteSchedules: vi.fn(),
+          getState: vi.fn(async () => ({ status: 'stopped' as const, lastChange: Date.now() })),
+          schedule: vi.fn() as Container['schedule'],
+        },
+        {
+          client,
+          storage,
+          stopOnStoppedState: false,
+          stoppedStateGraceSeconds: 900,
+          enforceBudgetStop: vi.fn(),
+        }
+      );
+
+      now.mockReturnValue(10_000);
+      await controller.billingHeartbeatTick();
+      expect(recordStop).not.toHaveBeenCalled();
+      expect((await getBillingContext(storage))?.stoppedObservedAtMs).toBe(10_000);
+
+      now.mockReturnValue(910_000);
+      await controller.billingHeartbeatTick();
+
+      expect(recordStop).toHaveBeenCalledWith(
+        expect.objectContaining({ reason: 'runtime_signal', usageSinceLast: 9 })
+      );
+      expect(await getBillingContext(storage)).toBeUndefined();
+    } finally {
+      now.mockRestore();
+    }
+  });
+
+  it('abandons local stopped-state retries after the hard ceiling', async () => {
+    const now = vi.spyOn(Date, 'now').mockReturnValue(1_000);
+    try {
+      const storage = memoryStorage();
+      await storedContext(storage);
+      const recordStop = vi.fn<ContainerUsageRpcMethods['recordStop']>(async () => {
+        throw new Error('meter unavailable');
+      });
+      const client = new ContainerUsageClient(
+        {
+          recordStart: async () => ({
+            success: true,
+            ack: { intervalId: 'instance-1:123', durable: 'pg', dedup: false },
+          }),
+          recordHeartbeat: async () => ({
+            intervalId: 'instance-1:123',
+            durable: 'pg',
+            dedup: false,
+            budget: { verdict: 'continue' },
+          }),
+          recordStop,
+        },
+        { service: 'cloud-agent-next', retry: { attempts: 1 } }
+      );
+      const deleteSchedules = vi.fn();
+      const controller = installBillingHeartbeat(
+        {
+          deleteSchedules,
+          getState: vi.fn(async () => ({ status: 'stopped' as const, lastChange: Date.now() })),
+          schedule: vi.fn() as Container['schedule'],
+        },
+        {
+          client,
+          storage,
+          stopOnStoppedState: false,
+          stoppedStateGraceSeconds: 900,
+          stoppedStateAbandonSeconds: 3_600,
+          enforceBudgetStop: vi.fn(),
+        }
+      );
+
+      now.mockReturnValue(10_000);
+      await controller.billingHeartbeatTick();
+      now.mockReturnValue(3_610_000);
+      await controller.billingHeartbeatTick();
+
+      expect(recordStop).toHaveBeenCalledOnce();
+      expect(await getBillingContext(storage)).toBeUndefined();
+      expect(deleteSchedules).toHaveBeenCalledWith(BILLING_HEARTBEAT_CALLBACK);
+    } finally {
+      now.mockRestore();
+    }
+  });
+
+  it('runs stop-delivery prerequisites before retrying a persisted stop', async () => {
+    const storage = memoryStorage();
+    await storedContext(storage);
+    const beforeStopDelivery = vi.fn(async () => undefined);
+    const recordStop = vi.fn<ContainerUsageRpcMethods['recordStop']>(async input => ({
+      intervalId: `${input.instanceId}:${input.startEpochMs}`,
+      durable: 'pg',
+      dedup: false,
+    }));
+    const client = new ContainerUsageClient(
+      {
+        recordStart: async () => ({
+          success: true,
+          ack: { intervalId: 'instance-1:123', durable: 'pg', dedup: false },
+        }),
+        recordHeartbeat: async () => ({
+          intervalId: 'instance-1:123',
+          durable: 'pg',
+          dedup: false,
+          budget: { verdict: 'continue' },
+        }),
+        recordStop,
+      },
+      { service: 'cloud-agent-next' }
+    );
+    const controller = installBillingHeartbeat(
+      {
+        deleteSchedules: vi.fn(),
+        getState: vi.fn(async () => ({ status: 'running' as const, lastChange: Date.now() })),
+        schedule: vi.fn() as Container['schedule'],
+      },
+      { client, storage, beforeStopDelivery, enforceBudgetStop: vi.fn() }
+    );
+    await controller.persistStop({ reason: 'exit', exitCode: 9 });
+
+    await controller.billingHeartbeatTick();
+
+    expect(beforeStopDelivery).toHaveBeenCalledOnce();
+    expect(beforeStopDelivery.mock.invocationCallOrder[0]).toBeLessThan(
+      recordStop.mock.invocationCallOrder[0] ?? Number.POSITIVE_INFINITY
+    );
+  });
+
   it('immediately retries an unacknowledged heartbeat with the same segment payload', async () => {
     const storage = memoryStorage();
     await storedContext(storage);