refactor: centralize schema resolution, create connections table during scaffolding

- Add getAppSchema() that reads DATABASE_SCHEMA from .env (single source of truth)
- Replace all ad-hoc schema lookups (dev-server, MCP tool, CLI) with getAppSchema()
- Move ensureConnectionsTable call into setupAppSchema (scaffolding) only
- Remove redundant ensureConnectionsTable from factory.ts and dev-server.ts
- Remove ensureConnectionsTable from public API exports and factory test mock

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: cevian

packages/core/src/__tests__/factory.test.ts

@@ -5,11 +5,6 @@ import { create0pflow } from "../factory.js";
 import { Workflow } from "../workflow.js";
 import { Node } from "../node.js";
 
-// Mock connections schema (avoids pg connection in tests)
-vi.mock("../connections/schema.js", () => ({
-  ensureConnectionsTable: vi.fn().mockResolvedValue(undefined),
-}));
-
 // Mock DBOS
 vi.mock("@dbos-inc/dbos-sdk", () => ({
   DBOS: {

packages/core/src/cli/app.ts

@@ -1,6 +1,7 @@
 // packages/cli/src/app.ts
 import fs from "fs";
 import path from "path";
+import * as dotenv from "dotenv";
 
 /**
  * Get the app name from package.json in cwd
@@ -20,3 +21,36 @@ export function getAppName(): string | undefined {
     return undefined;
   }
 }
+
+/**
+ * Get the app schema name from DATABASE_SCHEMA in the project's .env file.
+ * This is the canonical source — written by setup_app_schema.
+ * @param projectRoot Directory containing the .env file (defaults to cwd)
+ */
+export function getAppSchema(projectRoot?: string): string {
+  const envPath = path.join(projectRoot ?? process.cwd(), ".env");
+
+  if (!fs.existsSync(envPath)) {
+    throw new Error(
+      "DATABASE_SCHEMA not found in .env file. " +
+      "Run setup_app_schema to configure the database.",
+    );
+  }
+
+  try {
+    const env = dotenv.parse(fs.readFileSync(envPath, "utf-8"));
+    if (!env.DATABASE_SCHEMA) {
+      throw new Error(
+        "DATABASE_SCHEMA not found in .env file. " +
+        "Run setup_app_schema to configure the database.",
+      );
+    }
+    return env.DATABASE_SCHEMA;
+  } catch (err) {
+    if (err instanceof Error && err.message.includes("DATABASE_SCHEMA")) throw err;
+    throw new Error(
+      "DATABASE_SCHEMA not found in .env file. " +
+      "Run setup_app_schema to configure the database.",
+    );
+  }
+}

packages/core/src/cli/index.ts

@@ -12,7 +12,7 @@ import { discoverWorkflows, discoverNodes } from "./discovery.js";
 import { resolveEnv } from "./env.js";
 import { listRuns, getRun } from "./runs.js";
 import { getTrace, printTrace } from "./trace.js";
-import { getAppName } from "./app.js";
+import { getAppName, getAppSchema } from "./app.js";
 import { startMcpServer } from "./mcp/server.js";
 import { runInstall, runUninstall } from "./install.js";
 import { runRun } from "./run.js";
@@ -209,7 +209,7 @@ workflow
 
       const pflow = await create0pflow({
         databaseUrl: process.env.DATABASE_URL!,
-        appName: getAppName() ?? "opflow",
+        appName: getAppSchema(),
         workflows: workflowRegistry,
         nodes,
       });
@@ -347,7 +347,7 @@ node
 
       const pflow = await create0pflow({
         databaseUrl: process.env.DATABASE_URL!,
-        appName: getAppName() ?? "opflow",
+        appName: getAppSchema(),
         nodes,
       });
 

packages/core/src/cli/mcp/lib/scaffolding.ts

@@ -7,6 +7,7 @@ import * as dotenv from "dotenv";
 import pg from "pg";
 import { packageRoot, version } from "../config.js";
 import { writeAppTemplates, create0pflowDirectories } from "../lib/templates.js";
+import { ensureConnectionsTable } from "../../../connections/schema.js";
 
 const execAsync = (cmd: string, cwd?: string) =>
   new Promise<{ stdout: string; stderr: string }>((resolve, reject) => {
@@ -319,6 +320,9 @@ export async function setupAppSchema({
       .join("\n");
 
     await writeFile(envPath, `${newEnvContent}\n`);
+
+    // Create the opflow_connections table so it's ready before the dev UI launches
+    await ensureConnectionsTable(appDatabaseUrl, appName);
   } catch (err) {
     const error = err as Error;
     return {

packages/core/src/cli/mcp/tools/getConnectionInfo.ts

@@ -7,7 +7,7 @@ import pg from "pg";
 import type { ServerContext } from "../types.js";
 import { createIntegrationProvider } from "../../../connections/integration-provider.js";
 import { resolveConnectionId } from "../../../connections/resolver.js";
-import { getAppName } from "../../app.js";
+import { getAppSchema } from "../../app.js";
 
 const inputSchema = {
   integration_id: z
@@ -100,7 +100,7 @@ export const getConnectionInfoFactory: ApiFactory<
 
       // Look up connection_id using the same resolution as runtime:
       // exact (workflow_name, node_name) match first, then global (* / *) fallback
-      const appSchema = env.DATABASE_SCHEMA ?? getAppName();
+      const appSchema = getAppSchema();
       let connectionId: string | null = null;
       const pool = new pg.Pool({ connectionString: databaseUrl, max: 1 });
       try {

packages/core/src/dev-ui/dev-server.ts

@@ -7,9 +7,9 @@ import { createWSServer, type WSClientMessage } from "./ws.js";
 import { createWatcher } from "./watcher.js";
 import type { PtyManager } from "./pty.js";
 import { handleApiRequest } from "./api.js";
-import { ensureConnectionsTable } from "../connections/index.js";
 import { createIntegrationProvider } from "../connections/integration-provider.js";
 import { getSchemaName } from "../dbos.js";
+import { getAppSchema } from "../cli/app.js";
 import pg from "pg";
 
 const MIME: Record<string, string> = {
@@ -49,20 +49,21 @@ export async function startDevServer(options: DevServerOptions) {
   }
   const clientDir = resolve(pkgRoot, "dist/dev-ui-client");
 
-  // Resolve DBOS schema name from project's package.json
+  // Read app schema from project's .env (DATABASE_SCHEMA, written by setup_app_schema)
   const projectPkgPath = resolve(projectRoot, "package.json");
-  let appName: string | undefined;
+  let pkgName: string | undefined;
   try {
-    appName = JSON.parse(readFileSync(projectPkgPath, "utf-8")).name;
+    pkgName = JSON.parse(readFileSync(projectPkgPath, "utf-8")).name;
   } catch { /* use default */ }
-  const dbosSchema = getSchemaName(appName);
+
+  const appSchema = getAppSchema(projectRoot);
+  const dbosSchema = getSchemaName(pkgName);
 
   // Set up API context if database is configured
   const hasApi = !!(options.databaseUrl);
   let pool: pg.Pool | null = null;
 
   if (hasApi) {
-    await ensureConnectionsTable(options.databaseUrl!, appName ?? "opflow");
     pool = new pg.Pool({ connectionString: options.databaseUrl! });
   }
 
@@ -82,7 +83,7 @@ export async function startDevServer(options: DevServerOptions) {
           pool,
           integrationProvider: integrationProvider!,
           schema: dbosSchema,
-          appSchema: appName ?? "opflow",
+          appSchema,
         });
         if (handled) return;
       } catch (err) {

packages/core/src/factory.ts

@@ -5,7 +5,6 @@ import { initializeDBOS, shutdownDBOS } from "./dbos.js";
 import { NodeRegistry } from "./nodes/registry.js";
 import { configureAgentRuntime } from "./agent.js";
 import { Workflow, configureWorkflowRuntime, type NodeWrapper } from "./workflow.js";
-import { ensureConnectionsTable } from "./connections/index.js";
 import { createIntegrationProvider } from "./connections/integration-provider.js";
 import pg from "pg";
 
@@ -43,7 +42,6 @@ export async function create0pflow(config: PflowConfig): Promise<Pflow> {
   // Create shared pg pool for connection management (needed for local connection mapping)
   const pool = new pg.Pool({ connectionString: config.databaseUrl });
   const appSchema = config.appName;
-  await ensureConnectionsTable(config.databaseUrl, appSchema);
 
   // Configure workflow runtime with pool + integration provider
   configureWorkflowRuntime(pool, integrationProvider, appSchema);

packages/core/src/index.ts

@@ -49,7 +49,6 @@ export type {
 
 // Connection management
 export {
-  ensureConnectionsTable,
   resolveConnectionId,
   upsertConnection,
   listConnections,