fix(cloud-agent-next): isolate dev reporting and track activity

Author: eshurakov

services/cloud-agent-next/README.md

@@ -703,7 +703,7 @@ Required secrets:
 
 - `CLOUDFLARE_API_TOKEN` — must have permission to deploy the worker for the selected environment.
 
-Cloud Agent Next owns its operational reporting projection. Deploy the reporting database migration before this worker release and provision `cloud-agent-next-report-queue` plus `cloud-agent-next-report-queue-dlq` with four-day message retention. New sessions synchronously create their reporting anchor before setup; queued run reports do not synthesize parents when an anchor is absent.
+Cloud Agent Next owns its operational reporting projection. Deploy the reporting database migration before this worker release and provision `cloud-agent-next-report-queue` plus `cloud-agent-next-report-queue-dlq` with four-day message retention. A deployed `dev` environment requires isolated `cloud-agent-next-report-queue-dev` and `cloud-agent-next-report-queue-dlq-dev` resources with the same retention. New sessions synchronously create their reporting anchor before setup; queued run reports do not synthesize parents when an anchor is absent.
 
 ### Testing
 

services/cloud-agent-next/src/persistence/CloudAgentSession.ts

@@ -596,6 +596,7 @@ export class CloudAgentSession extends DurableObject<WorkerEnv> {
         keepContainerAlive: () => {
           void this.keepContainerAlive();
         },
+        observeCorrelatedAgentActivity: messageId => this.recordCorrelatedAgentActivity(messageId),
         terminalizeSessionMessageOnce: async (messageId, params, wrapperRunId) => {
           await this.ensureAcceptedMessageBeforeTerminal(messageId, wrapperRunId);
           await this.recordCorrelatedAgentActivity(messageId);
@@ -2051,7 +2052,7 @@ export class CloudAgentSession extends DurableObject<WorkerEnv> {
 
   private async recordCorrelatedAgentActivity(messageId: string): Promise<void> {
     const updated = await markAgentActivityObserved(this.ctx.storage, messageId);
-    if (updated) void this.reportRunState(updated).catch(() => undefined);
+    if (updated) this.ctx.waitUntil(this.reportRunState(updated));
   }
 
   private async ensureAcceptedMessageBeforeTerminal(

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

@@ -51,6 +51,7 @@ vi.mock('./callbacks/index.js', () => ({
 vi.mock('./telemetry/report-consumer.js', () => ({
   CLOUD_AGENT_REPORT_QUEUE_NAMES: new Set([
     'cloud-agent-next-report-queue',
+    'cloud-agent-next-report-queue-dev',
     'cloud-agent-next-report-queue-test',
   ]),
   consumeCloudAgentReportBatch: consumeCloudAgentReportBatchMock,
@@ -160,6 +161,18 @@ describe('server background reporting', () => {
     expect(consumeCloudAgentReportBatchMock).toHaveBeenCalledWith(batch, env);
   });
 
+  it('routes isolated development report queue batches to the Cloud Agent report consumer', async () => {
+    const env = createEnv();
+    const batch = {
+      queue: 'cloud-agent-next-report-queue-dev',
+      messages: [],
+    } as unknown as MessageBatch<unknown>;
+
+    await worker.queue(batch, env as unknown as Env);
+
+    expect(consumeCloudAgentReportBatchMock).toHaveBeenCalledWith(batch, env);
+  });
+
   it('runs reporting retention cleanup from the scheduled handler', async () => {
     const env = createEnv();
 

services/cloud-agent-next/src/telemetry/report-consumer.test.ts

@@ -9,7 +9,7 @@ vi.mock('./report-store.js', () => ({
 }));
 
 import { createCloudAgentReportStore } from './report-store.js';
-import { consumeCloudAgentReportBatch } from './report-consumer.js';
+import { CLOUD_AGENT_REPORT_QUEUE_NAMES, consumeCloudAgentReportBatch } from './report-consumer.js';
 
 const report = {
   version: 1,
@@ -38,6 +38,10 @@ describe('Cloud Agent report consumer', () => {
     vi.resetAllMocks();
   });
 
+  it('routes isolated development reporting queue messages', () => {
+    expect(CLOUD_AGENT_REPORT_QUEUE_NAMES.has('cloud-agent-next-report-queue-dev')).toBe(true);
+  });
+
   it('acks a valid saved report after its essential write completes', async () => {
     const saveReport = vi.fn(async () => ({ outcome: 'applied' as const }));
     vi.mocked(createCloudAgentReportStore).mockReturnValue({ saveReport } as never);

services/cloud-agent-next/src/telemetry/report-consumer.ts

@@ -6,6 +6,7 @@ import { createCloudAgentReportStore } from './report-store.js';
 
 export const CLOUD_AGENT_REPORT_QUEUE_NAMES = new Set([
   'cloud-agent-next-report-queue',
+  'cloud-agent-next-report-queue-dev',
   'cloud-agent-next-report-queue-test',
 ]);
 

services/cloud-agent-next/src/websocket/ingest.test.ts

@@ -847,11 +847,12 @@ describe('createIngestHandler', () => {
     function createNewPathDOContext() {
       return {
         ...createFakeDOContext(),
+        observeCorrelatedAgentActivity: vi.fn().mockResolvedValue(undefined),
         terminalizeSessionMessageOnce: vi.fn().mockResolvedValue(undefined),
       };
     }
 
-    it('does NOT terminalize on partial assistant message.updated (no time.completed)', async () => {
+    it('observes activity without terminalizing on partial assistant message.updated', async () => {
       const state = createFakeState();
       const doContext = createNewPathDOContext();
       const handler = createIngestHandler(
@@ -882,6 +883,7 @@ describe('createIngestHandler', () => {
 
       await handler.handleIngestMessage(ws, message);
 
+      expect(doContext.observeCorrelatedAgentActivity).toHaveBeenCalledWith('msg_user_111');
       expect(doContext.terminalizeSessionMessageOnce).not.toHaveBeenCalled();
     });
 

services/cloud-agent-next/src/websocket/ingest.ts

@@ -144,6 +144,7 @@ export type IngestDOContext = {
     | 'onTerminalEvent'
   >;
   keepContainerAlive?: () => void;
+  observeCorrelatedAgentActivity?: (messageId: string) => Promise<void>;
   terminalizeSessionMessageOnce?: (
     messageId: string,
     params: {
@@ -609,18 +610,25 @@ export function createIngestHandler(
             const assistantError = getAssistantErrorMessage(info?.error);
             const hasError = assistantError !== undefined;
             const isTerminal = isCompleted || hasError;
-            if (info?.role === 'assistant' && typeof info.parentID === 'string' && isTerminal) {
-              await doContext.terminalizeSessionMessageOnce?.(
-                info.parentID,
-                {
-                  kind: hasError ? 'failed' : 'completed',
-                  assistantMessageId: typeof info.id === 'string' ? info.id : undefined,
-                  completionSource: 'assistant_message_event',
-                  reason: hasError ? 'assistant_error' : undefined,
-                  error: assistantError,
-                },
-                wrapperRunId
-              );
+            const parentMessageId =
+              info?.role === 'assistant' && typeof info.parentID === 'string'
+                ? info.parentID
+                : undefined;
+            if (parentMessageId !== undefined) {
+              await doContext.observeCorrelatedAgentActivity?.(parentMessageId);
+              if (isTerminal) {
+                await doContext.terminalizeSessionMessageOnce?.(
+                  parentMessageId,
+                  {
+                    kind: hasError ? 'failed' : 'completed',
+                    assistantMessageId: typeof info?.id === 'string' ? info.id : undefined,
+                    completionSource: 'assistant_message_event',
+                    reason: hasError ? 'assistant_error' : undefined,
+                    error: assistantError,
+                  },
+                  wrapperRunId
+                );
+              }
             }
           }
         }

services/cloud-agent-next/wrangler.jsonc

@@ -363,7 +363,7 @@
           },
           {
             "binding": "CLOUD_AGENT_REPORT_QUEUE",
-            "queue": "cloud-agent-next-report-queue",
+            "queue": "cloud-agent-next-report-queue-dev",
           },
         ],
         "consumers": [
@@ -374,9 +374,9 @@
             "max_retries": 4,
           },
           {
-            "queue": "cloud-agent-next-report-queue",
+            "queue": "cloud-agent-next-report-queue-dev",
             "max_retries": 3,
-            "dead_letter_queue": "cloud-agent-next-report-queue-dlq",
+            "dead_letter_queue": "cloud-agent-next-report-queue-dlq-dev",
           },
         ],
       },