fix(cloud): harden mock stack e2e harness

Author: NubsCarson

packages/cloud-api/v1/_container-control-plane-forward.ts

@@ -51,15 +51,19 @@ async function forwardControlPlaneRequest(
   configureHeaders(headers);
 
   try {
-    const upstream = await fetch(target, {
-      body:
-        c.req.method === "GET" || c.req.method === "HEAD"
-          ? undefined
-          : c.req.raw.body,
+    const body =
+      c.req.method === "GET" || c.req.method === "HEAD"
+        ? undefined
+        : c.req.raw.body;
+    const init: RequestInit & { duplex?: "half" } = {
+      body,
       headers,
       method: c.req.method,
       redirect: "manual",
-    });
+    };
+    if (body) init.duplex = "half";
+
+    const upstream = await fetch(target, init);
 
     return new Response(upstream.body, {
       headers: upstream.headers,

packages/cloud-api/v1/eliza/agents/[agentId]/resume/route.ts

@@ -245,12 +245,8 @@ async function __hono_POST(
 const __hono_app = new Hono<AppEnv>();
 __hono_app.options("/", () => handleCorsOptions(CORS_METHODS));
 __hono_app.post("/", async (c) =>
-  __hono_POST(
-    c.req.raw,
-    c.env,
-    {
-      params: Promise.resolve({ agentId: c.req.param("agentId")! }),
-    },
-  ),
+  __hono_POST(c.req.raw, c.env, {
+    params: Promise.resolve({ agentId: c.req.param("agentId")! }),
+  }),
 );
 export default __hono_app;

packages/cloud-api/v1/eliza/agents/[agentId]/snapshot/route.ts

@@ -96,12 +96,8 @@ async function __hono_POST(
 const __hono_app = new Hono<AppEnv>();
 __hono_app.options("/", () => handleCorsOptions(CORS_METHODS));
 __hono_app.post("/", async (c) =>
-  __hono_POST(
-    c.req.raw,
-    c.env,
-    {
-      params: Promise.resolve({ agentId: c.req.param("agentId")! }),
-    },
-  ),
+  __hono_POST(c.req.raw, c.env, {
+    params: Promise.resolve({ agentId: c.req.param("agentId")! }),
+  }),
 );
 export default __hono_app;

packages/cloud-api/v1/eliza/agents/[agentId]/suspend/route.ts

@@ -131,12 +131,8 @@ async function __hono_POST(
 const __hono_app = new Hono<AppEnv>();
 __hono_app.options("/", () => handleCorsOptions(CORS_METHODS));
 __hono_app.post("/", async (c) =>
-  __hono_POST(
-    c.req.raw,
-    c.env,
-    {
-      params: Promise.resolve({ agentId: c.req.param("agentId")! }),
-    },
-  ),
+  __hono_POST(c.req.raw, c.env, {
+    params: Promise.resolve({ agentId: c.req.param("agentId")! }),
+  }),
 );
 export default __hono_app;

packages/cloud-api/webhooks/bluebubbles/route.test.ts

@@ -1,4 +1,5 @@
 import { beforeEach, describe, expect, mock, test } from "bun:test";
+import type { RegisterPhoneGatewayDeviceResult } from "@/lib/services/phone-gateway-devices";
 
 const routePhoneMessage = mock(async () => ({
   handled: true,
@@ -7,10 +8,12 @@ const routePhoneMessage = mock(async () => ({
   userId: "user-1",
   organizationId: "org-1",
 }));
-const registerPhoneGatewayDevice = mock(async () => ({
-  id: "gateway-device-1",
-  registered: true,
-}));
+const registerPhoneGatewayDevice = mock(
+  async (): Promise<RegisterPhoneGatewayDeviceResult> => ({
+    id: "gateway-device-1",
+    registered: true,
+  }),
+);
 
 mock.module("@/lib/services/agent-gateway-router", () => ({
   agentGatewayRouterService: {

packages/cloud-shared/src/db/client.ts

@@ -34,6 +34,11 @@ type Database = NodePgDatabase<typeof schema>;
 /** Transaction handle for `writeTransaction` callbacks. */
 type DbTransaction = NodePgTransaction<typeof schema, SchemaTables>;
 
+interface ManagedConnection {
+  database: Database;
+  close?: () => Promise<void>;
+}
+
 /**
  * Get the primary database URL (always required)
  */
@@ -81,7 +86,7 @@ function parsePGliteDataDir(url: string): string {
  * call site type as `Database`; PGlite is bun/node-only and does not exist
  * on the Workers runtime.
  */
-function createPGliteClient(dataDir: string): Database {
+function createPGliteClient(dataDir: string): ManagedConnection {
   const { PGlite } = require("@electric-sql/pglite") as typeof import("@electric-sql/pglite");
   const { vector } =
     require("@electric-sql/pglite/vector") as typeof import("@electric-sql/pglite/vector");
@@ -90,7 +95,12 @@ function createPGliteClient(dataDir: string): Database {
     extensions: { vector },
   });
   const database: PgliteDatabase<typeof schema> = drizzlePGlite({ client, schema });
-  return database as Database;
+  return {
+    database: database as Database,
+    close: async () => {
+      await client.close();
+    },
+  };
 }
 
 function isLocalTcpPostgresUrl(url: string): boolean {
@@ -149,7 +159,7 @@ function createPgPool(url: string): PgPool {
 /**
  * Create a database connection from a URL
  */
-function createConnection(url: string): Database {
+function createConnection(url: string): ManagedConnection {
   if (url.startsWith("pglite://")) {
     if (isCloudflareWorkerRuntime()) {
       throw new Error("pglite:// URLs are local-only and cannot run inside a Cloudflare Worker.");
@@ -164,11 +174,21 @@ function createConnection(url: string): Database {
       neonConfig.webSocketConstructor = ws;
     }
     const pool = new NeonPool({ connectionString: url });
-    return drizzleNeon(pool, { schema }) as Database;
+    return {
+      database: drizzleNeon(pool, { schema }) as Database,
+      close: async () => {
+        await pool.end();
+      },
+    };
   }
 
   const pool = createPgPool(url);
-  return drizzleNode(pool, { schema }) as Database;
+  return {
+    database: drizzleNode(pool, { schema }) as Database,
+    close: async () => {
+      await pool.end();
+    },
+  };
 }
 
 // ============================================================================
@@ -207,7 +227,7 @@ export async function runWithDbCacheAsync<T>(fn: () => Promise<T>): Promise<T> {
  * can safely live for the lifetime of the process.
  */
 class DatabaseConnectionManager {
-  private connections: Map<string, Database> = new Map();
+  private connections: Map<string, ManagedConnection> = new Map();
   private initialized = false;
 
   /**
@@ -224,18 +244,18 @@ class DatabaseConnectionManager {
       if (requestCache) {
         let cached = requestCache.get(url);
         if (!cached) {
-          cached = createConnection(url);
+          cached = createConnection(url).database;
           requestCache.set(url, cached);
         }
         return cached;
       }
-      return createConnection(url);
+      return createConnection(url).database;
     }
 
     if (!this.connections.has(url)) {
       this.connections.set(url, createConnection(url));
     }
-    return this.connections.get(url)!;
+    return this.connections.get(url)!.database;
   }
 
   /**
@@ -265,6 +285,16 @@ class DatabaseConnectionManager {
       databaseUrlConfigured: !!applyDatabaseUrlFallback(env),
     };
   }
+
+  async closeAll(): Promise<void> {
+    const connections = [...this.connections.values()];
+    this.connections.clear();
+    await Promise.all(
+      connections.map(async (connection) => {
+        await connection.close?.();
+      }),
+    );
+  }
 }
 
 const connectionManager = new DatabaseConnectionManager();
@@ -329,6 +359,14 @@ export function getDbConnectionInfo() {
   return connectionManager.getConnectionInfo();
 }
 
+/**
+ * Close process-level DB pools. Mainly used by local/E2E harnesses before
+ * shutting down an ephemeral database server.
+ */
+export async function closeDbConnections(): Promise<void> {
+  await connectionManager.closeAll();
+}
+
 /**
  * Execute a read-intent query.
  */

packages/cloud-shared/src/lib/services/api-keys.ts

@@ -254,11 +254,10 @@ export class ApiKeysService {
 
   async revokeForAgent(agentSandboxId: string): Promise<void> {
     const name = ApiKeysService.agentApiKeyName(agentSandboxId);
-    const keys = await apiKeysRepository.findByName(name);
+    const keys = await apiKeysRepository.deleteByName(name);
     for (const key of keys) {
       await this.invalidateCache(key.key_hash);
     }
-    await apiKeysRepository.deleteByName(name);
   }
 }
 

packages/cloud-shared/src/lib/services/eliza-sandbox.ts

@@ -516,7 +516,7 @@ export class ElizaSandboxService {
   }
 
   async deleteAgent(agentId: string, orgId: string): Promise<DeleteAgentResult> {
-    return dbWrite.transaction(async (tx) => {
+    const result = await dbWrite.transaction(async (tx) => {
       await this.lockLifecycle(tx, agentId, orgId);
 
       const rec = await this.getAgentForLifecycleMutation(tx, agentId, orgId);
@@ -584,23 +584,26 @@ export class ElizaSandboxService {
       `);
       const deletedSandbox = result.rows[0];
 
-      if (deletedSandbox) {
-        // Best-effort: revoke the per-agent API key. A failure here doesn't
-        // un-delete the sandbox; the key just lingers as inactive data.
-        try {
-          await apiKeysService.revokeForAgent(agentId);
-        } catch (err) {
-          logger.warn("[agent-sandbox] Failed to revoke per-agent API key", {
-            agentId,
-            error: err instanceof Error ? err.message : String(err),
-          });
-        }
-      }
-
       return deletedSandbox
         ? ({ success: true, deletedSandbox } as const)
         : ({ success: false, error: "Agent not found" } as const);
     });
+
+    if (result.success) {
+      // Best-effort: revoke the per-agent API key after the row delete commits.
+      // A failure here does not un-delete the sandbox; the key just lingers as
+      // inactive data and can be cleaned by ops.
+      try {
+        await apiKeysService.revokeForAgent(agentId);
+      } catch (err) {
+        logger.warn("[agent-sandbox] Failed to revoke per-agent API key", {
+          agentId,
+          error: err instanceof Error ? err.message : String(err),
+        });
+      }
+    }
+
+    return result;
   }
 
   /**