fix(datatables): never grant CREATEROLE to custom_instance_user on the external cluster

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

CONTEXT.md

@@ -37,6 +37,25 @@ _Avoid_: argument field, param
 Any other place a property can be picked into: the loop iterator, skip and early-stop predicates, the retry condition, a branch predicate, timeout. Its prop picker opens in a popover from the connect button rather than taking a pane.
 _Avoid_: JS field, code input
 
+### Flow chat
+
+**Conversation**:
+One thread of messages against one chat-enabled flow, with its own agent memory. A flow has
+many; the chat shows one at a time.
+_Avoid_: thread, session (that names an AI session, a different thing), chat (that names the surface)
+
+**Turn**:
+One question and the answer to it: the run the question started, the handle that stops it,
+and the rows it is writing. At most one per conversation, and the chat is held for its whole
+length — from the moment the question takes the chat, before it has a job, until it is ended.
+_Avoid_: request, exchange, message round
+
+**Transcript**:
+The rows a conversation's chat holds. Not the conversation: it is the newest page plus
+whatever older pages the reader has scrolled back through, so a question it cannot answer
+from what it holds is one to ask the server rather than to guess at.
+_Avoid_: history, messages (too easily read as "all of them")
+
 ### Permissions
 
 **Member**:

ai_evals/adapters/frontend/mockBackend.ts

@@ -92,7 +92,7 @@ export interface BenchmarkWorkspaceResource {
 }
 
 export interface BenchmarkWorkspaceJob {
-	/** Stable id so a case prompt can reference a specific run (e.g. for get_job_logs). */
+	/** Stable id so a case prompt can reference a specific run (e.g. for get_run). */
 	id?: string
 	jobKind?: CompletedJob['job_kind']
 	scriptPath?: string
@@ -100,6 +100,8 @@ export interface BenchmarkWorkspaceJob {
 	label?: string
 	success?: boolean
 	logs?: string
+	args?: Record<string, unknown>
+	result?: unknown
 }
 
 export interface BenchmarkWorkspaceRunnables {
@@ -156,7 +158,7 @@ export function registerBenchmarkWorkspaceRunnables(
 		...runnables,
 		datatables: runnables.datatables ? structuredClone(runnables.datatables) : undefined
 	})
-	// Seed any fixture jobs so list_runs / get_job_logs have data to return.
+	// Seed any fixture jobs so list_runs / get_run have data to return.
 	for (const seed of runnables.jobs ?? []) {
 		createBenchmarkCompletedJob({
 			workspace,
@@ -166,7 +168,9 @@ export function registerBenchmarkWorkspaceRunnables(
 			scriptPath: seed.scriptPath,
 			createdBy: seed.createdBy,
 			label: seed.label,
-			logs: seed.logs
+			logs: seed.logs,
+			args: seed.args,
+			result: seed.result
 		})
 	}
 }
@@ -481,6 +485,33 @@ export function getBenchmarkJobLogs(workspace: string, jobId: string): string {
 	return job.logs ?? ''
 }
 
+/**
+ * Mirror `JobService.getFlowAllResults`, which get_run calls for the execution
+ * tree. Fixture jobs are single runs with no steps, so only the root entry.
+ */
+export function getBenchmarkFlowAllResults(workspace: string, jobId: string) {
+	const job = getBenchmarkCompletedJob(workspace, jobId)
+	if (!job) {
+		throw new Error(`Job "${jobId}" not found in benchmark workspace`)
+	}
+	return {
+		entries: [
+			{
+				job_id: jobId,
+				label: 'Flow',
+				kind: job.job_kind ?? 'script',
+				depth: 0,
+				sibling_index: 1,
+				sibling_count: 1,
+				status: job.success ? 'success' : 'failure',
+				success: job.success
+			}
+		],
+		truncated: false,
+		scope_filtered: false
+	}
+}
+
 // ============= Drafts (per-user, DB-backed in production) =============
 
 /**

ai_evals/adapters/frontend/vitestAdapter.test.ts

@@ -62,6 +62,7 @@ vi.mock('$lib/gen', async () => {
 		getBenchmarkDatatableSchema,
 		getBenchmarkDraftForUser,
 		getBenchmarkFlowByPath,
+		getBenchmarkFlowAllResults,
 		getBenchmarkJobLogs,
 		getBenchmarkOwnDraft,
 		getBenchmarkScriptByHash,
@@ -325,7 +326,11 @@ vi.mock('$lib/gen', async () => {
 			getJobLogs: async (data: { workspace: string; id: string }) =>
 				hasBenchmarkWorkspace(data.workspace)
 					? getBenchmarkJobLogs(data.workspace, data.id)
-					: actual.JobService.getJobLogs(data)
+					: actual.JobService.getJobLogs(data),
+			getFlowAllResults: async (data: { workspace: string; id: string }) =>
+				hasBenchmarkWorkspace(data.workspace)
+					? getBenchmarkFlowAllResults(data.workspace, data.id)
+					: actual.JobService.getFlowAllResults(data)
 		}),
 		WorkspaceService: wrapService(actual.WorkspaceService, {
 			getCopilotInfo: async (data: { workspace: string }) =>

ai_evals/cases/global.yaml

@@ -889,13 +889,13 @@
     draftCountExactly: 0
   toolExpect:
     requiredToolsUsed:
-      - get_job_logs
+      - get_run
     forbiddenToolsUsed:
       - deploy_workspace_item
       - delete_workspace_item
       - write_script
     toolCallArgs:
-      - tool: get_job_logs
+      - tool: get_run
         field: id
         stringIncludesAnyOf:
           - 01920000-0000-7000-8000-0000000000f1
@@ -906,6 +906,34 @@
     - fetches the logs for the requested job id
     - explains the failure from the returned logs (connection refused to the upstream API)
 
+- id: global-run-args-and-result
+  prompt: |-
+    What was the run 01920000-0000-7000-8000-0000000000f2 called with, and what did it return?
+  initial: ai_evals/fixtures/frontend/global/initial/jobs_seed.json
+  runtime:
+    maxTurns: 6
+  validate:
+    draftCountExactly: 0
+  toolExpect:
+    requiredToolsUsed:
+      - get_run
+    forbiddenToolsUsed:
+      - test_run_script
+      - run_script
+      - deploy_workspace_item
+    toolCallArgs:
+      - tool: get_run
+        field: id
+        stringIncludesAnyOf:
+          - 01920000-0000-7000-8000-0000000000f2
+  # Read-only, so no draft for the global judge to score — validated on tool use
+  # and the deterministic argument check, like the neighbouring run cases.
+  skipJudge: true
+  judgeChecklist:
+    - reports the arguments the run was called with (region emea, 12 recipients)
+    - reports what the run returned (12 sent, 3 skipped)
+    - does not start a new run to find out
+
 # --- Page navigation (open_page) ---
 # The assistant should take the user to a Windmill page (Runs/Schedules) with the
 # right filters via open_page, rather than describing where to click or dumping the

ai_evals/fixtures/frontend/global/initial/jobs_seed.json

@@ -15,6 +15,8 @@
         "jobKind": "script",
         "createdBy": "bob",
         "success": true,
+        "args": { "region": "emea", "dry_run": false, "recipients": 12 },
+        "result": { "sent": 12, "skipped": 3, "digest_url": "https://reports.example.com/d/2026-06-09" },
         "logs": "Generating daily digest...\nDigest emailed to 12 recipients\nDone in 1.2s"
       },
       {

backend/.sqlx/query-06abbf945bee93349ff88f64906b96ea1e853ef202510281427cfa9beeff81b3.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT 1 AS one FROM workspace_settings WHERE workspace_id = $1 FOR UPDATE",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "one",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "06abbf945bee93349ff88f64906b96ea1e853ef202510281427cfa9beeff81b3"
+}

backend/.sqlx/query-06ce02cd7ce2f5a57355153edb573c242f9ba758db66e9a5e16f30e3e1494201.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace_settings ws\n           SET datatable = (\n               SELECT jsonb_set(ws.datatable, '{datatables}', jsonb_object_agg(\n                   dt.key,\n                   CASE WHEN dt.value->'reference'->>'workspace_id' = $1\n                             AND dt.value->'reference'->>'datatable' = $2\n                       THEN jsonb_set(dt.value, '{reference,datatable}', to_jsonb($3::text))\n                       ELSE dt.value END\n               ))\n               FROM jsonb_each(ws.datatable->'datatables') dt\n           )\n           WHERE EXISTS (\n               SELECT 1 FROM jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) d\n               WHERE d.value->'reference'->>'workspace_id' = $1\n                 AND d.value->'reference'->>'datatable' = $2\n           )",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "06ce02cd7ce2f5a57355153edb573c242f9ba758db66e9a5e16f30e3e1494201"
+}

backend/.sqlx/query-07a005f0f9e80a156cd2a5a0ae39a1fabeaa167818206a25abfe31d5582f942a.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM flow_conversation c\n              WHERE c.id = ANY($1)\n                AND c.workspace_id = $2\n                AND NOT EXISTS (\n                    SELECT 1 FROM flow_conversation_message m WHERE m.conversation_id = c.id\n                )",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "UuidArray",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "07a005f0f9e80a156cd2a5a0ae39a1fabeaa167818206a25abfe31d5582f942a"
+}

backend/.sqlx/query-0d10e0fa5cf4033c7d93c9ed56be8209046007917f44da954eccf2188e5bff1f.json

@@ -0,0 +1,29 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT permissioned_as, permissioned_as_email FROM v2_job\n                 WHERE id = $1 AND workspace_id = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "permissioned_as",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "permissioned_as_email",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false
+    ]
+  },
+  "hash": "0d10e0fa5cf4033c7d93c9ed56be8209046007917f44da954eccf2188e5bff1f"
+}

backend/.sqlx/query-297c7a40dfce729d44aa37bc7c65560517bd25e40c0752a00467829191e2eb98.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace_settings ws\n           SET datatable = (\n               SELECT jsonb_set(ws.datatable, '{datatables}', jsonb_object_agg(\n                   dt.key,\n                   CASE WHEN dt.value->'reference'->>'workspace_id' = $2\n                       THEN jsonb_set(dt.value, '{reference,workspace_id}', to_jsonb($1::text))\n                       ELSE dt.value END\n               ))\n               FROM jsonb_each(ws.datatable->'datatables') dt\n           )\n           WHERE jsonb_typeof(ws.datatable->'datatables') = 'object'\n             AND ws.datatable::text LIKE '%\"reference\"%'",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "297c7a40dfce729d44aa37bc7c65560517bd25e40c0752a00467829191e2eb98"
+}

backend/.sqlx/query-2a391cc1bfcd2f75b46144a394c01237e09c3060da88170f1f6e06468309d213.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture_config SET server_id = NULL, last_server_ping = NULL\n             WHERE workspace_id = $1 AND trigger_kind = 'postgres'\n               AND (trigger_config->>'postgres_resource_path' = $2\n                    OR trigger_config->>'postgres_resource_path' LIKE $3)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "2a391cc1bfcd2f75b46144a394c01237e09c3060da88170f1f6e06468309d213"
+}

backend/.sqlx/query-334dbcd48fb59c96c62c2705ab2d1ce716cd52417f487cc1a8dd376017b2db7d.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = $1)",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "exists",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Name"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "334dbcd48fb59c96c62c2705ab2d1ce716cd52417f487cc1a8dd376017b2db7d"
+}

backend/.sqlx/query-462d2b2822b185a6f51fafcfa957cb3b31ee6b69abae79a214dddba0dee4425c.json

@@ -1,15 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM flow_conversation_message m\n         USING flow_conversation c\n         WHERE m.conversation_id = c.id AND c.workspace_id = $1 AND m.job_id = ANY($2)",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text",
-        "UuidArray"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "462d2b2822b185a6f51fafcfa957cb3b31ee6b69abae79a214dddba0dee4425c"
-}

backend/.sqlx/query-4692810d2be817bbb5de9b476d68d695941bd4fb5ccef393e4da522ed479d601.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE postgres_trigger SET server_id = NULL, last_server_ping = NULL\n         WHERE workspace_id = $1\n           AND (postgres_resource_path = $2 OR postgres_resource_path LIKE $3)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "4692810d2be817bbb5de9b476d68d695941bd4fb5ccef393e4da522ed479d601"
+}

backend/.sqlx/query-4f52bf546579f26a1d22c239b8b0054b753cfbeb5dad1e8120fd5e8a672d50ef.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id FROM flow_conversation WHERE id = ANY($1) AND workspace_id = $2 ORDER BY id FOR UPDATE",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "UuidArray",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "4f52bf546579f26a1d22c239b8b0054b753cfbeb5dad1e8120fd5e8a672d50ef"
+}

backend/.sqlx/query-5048e21546f9710697100100e1255ab103979433bc386d7c89d0e30db12bfd57.json

@@ -0,0 +1,29 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT ws.workspace_id AS \"workspace_id!\", dt.key AS \"datatable!\"\n           FROM workspace_settings ws\n           CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt\n           WHERE dt.value->'reference'->>'workspace_id' = $1\n             AND dt.value->'reference'->>'datatable' = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "datatable!",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      null
+    ]
+  },
+  "hash": "5048e21546f9710697100100e1255ab103979433bc386d7c89d0e30db12bfd57"
+}

backend/.sqlx/query-538dd1779874e4003932d7f17750239c625f85e25b3364bf2edf566f518c8ee2.json

@@ -0,0 +1,29 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT ws.workspace_id AS \"workspace_id!\", dt.key AS \"datatable!\"\n               FROM workspace_settings ws\n               CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt\n               WHERE dt.value->'reference'->>'workspace_id' = $1\n                 AND dt.value->'reference'->>'datatable' = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "datatable!",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      null
+    ]
+  },
+  "hash": "538dd1779874e4003932d7f17750239c625f85e25b3364bf2edf566f518c8ee2"
+}

backend/.sqlx/query-58e5cfe9eb87bda9f7de87c403861b6e7b9d35a41594681e2a92a87359e6a018.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT username FROM usr WHERE workspace_id = $1 AND email = $2 AND disabled = false",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "username",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "58e5cfe9eb87bda9f7de87c403861b6e7b9d35a41594681e2a92a87359e6a018"
+}

backend/.sqlx/query-69bfbe9b39414b724488532cc3b3659915d9fcb3d58f16532aaffe06c44ec976.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM flow_conversation_message m\n         USING flow_conversation c\n         WHERE m.conversation_id = c.id AND c.workspace_id = $1 AND m.job_id = ANY($2)\n         RETURNING m.conversation_id",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "conversation_id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "UuidArray"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "69bfbe9b39414b724488532cc3b3659915d9fcb3d58f16532aaffe06c44ec976"
+}

backend/.sqlx/query-6f32c1feed096ff706ae359ad6a3ca33b3f82ca38289dfa4a69aa95041027d57.json

@@ -0,0 +1,59 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, workspace_id, flow_path, title, created_at, updated_at, created_by\n         FROM flow_conversation\n         WHERE id = $1 AND workspace_id = $2\n         FOR UPDATE",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 1,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "flow_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "title",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 4,
+        "name": "created_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 5,
+        "name": "updated_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 6,
+        "name": "created_by",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      true,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "6f32c1feed096ff706ae359ad6a3ca33b3f82ca38289dfa4a69aa95041027d57"
+}

backend/.sqlx/query-6f9fb5d72f486358fa25d6887bd69b93910e028f140c07048f2c1c8d63ee6909.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO global_settings (name, value) VALUES ($1, $2)\n         ON CONFLICT (name) DO UPDATE SET value = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "6f9fb5d72f486358fa25d6887bd69b93910e028f140c07048f2c1c8d63ee6909"
+}

backend/.sqlx/query-71ee2cb6661cca1fa4d8874a7f6d368347c59f36fd87df6dc7996152ccb84af0.json

@@ -0,0 +1,38 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, name, enabled, pwd FROM datatable_role",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "name",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "enabled",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 3,
+        "name": "pwd",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      true
+    ]
+  },
+  "hash": "71ee2cb6661cca1fa4d8874a7f6d368347c59f36fd87df6dc7996152ccb84af0"
+}

backend/.sqlx/query-79799b5a2e499df6c28e286c42b9ad2db940c2455ab19cc95e5198baf96d5629.json

@@ -0,0 +1,29 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT ws.workspace_id AS \"workspace_id!\", dt.key AS \"datatable!\"\n        FROM workspace_settings ws\n        JOIN workspace w ON w.id = ws.workspace_id AND w.deleted = false\n        CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt\n        WHERE ws.workspace_id <> $1\n          AND dt.value->'database'->>'resource_type' = 'instance'\n          AND dt.value->'database'->>'resource_path' = $2\n        ORDER BY ws.workspace_id, dt.key\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "datatable!",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      null
+    ]
+  },
+  "hash": "79799b5a2e499df6c28e286c42b9ad2db940c2455ab19cc95e5198baf96d5629"
+}

backend/.sqlx/query-86af9d51a158ea5cb6161461ecddf2a63695f8cbf8af648da5a0a77a5b9d02ba.json

@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO datatable_role (id, name, enabled, pwd) VALUES ($1, $2, $3, $4)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Bool",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "86af9d51a158ea5cb6161461ecddf2a63695f8cbf8af648da5a0a77a5b9d02ba"
+}

backend/.sqlx/query-89ea81b765550cf665e30533efc9672f8c98d2579fc72c07752361cc5fd683dc.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM ai_agent_memory a\n               USING flow_conversation c\n              WHERE c.id = ANY($1)\n                AND c.workspace_id = $2\n                AND a.conversation_id = c.id\n                AND a.workspace_id = c.workspace_id\n                AND NOT EXISTS (\n                    SELECT 1 FROM flow_conversation_message m WHERE m.conversation_id = c.id\n                )",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "UuidArray",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "89ea81b765550cf665e30533efc9672f8c98d2579fc72c07752361cc5fd683dc"
+}

backend/.sqlx/query-90b910da8d00a7c7bcf29c167e38e44eb1c0062a8241dc8fe0ed3dd95b65f89a.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM flow_conversation c\n              WHERE c.id = ANY($1)\n                AND NOT EXISTS (\n                    SELECT 1 FROM flow_conversation_message m WHERE m.conversation_id = c.id\n                )",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "UuidArray"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "90b910da8d00a7c7bcf29c167e38e44eb1c0062a8241dc8fe0ed3dd95b65f89a"
+}

backend/.sqlx/query-967f52005f4a044b3a2e9f02ceadf90dad5246681bde6caaa633b85a5e8b2352.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM flow_conversation_message WHERE job_id = ANY($1) RETURNING conversation_id",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "conversation_id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "UuidArray"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "967f52005f4a044b3a2e9f02ceadf90dad5246681bde6caaa633b85a5e8b2352"
+}

backend/.sqlx/query-975e404ac3a6258bb8220e122e3de094c7ab23330fdbc74d6e4ad472ddd3c820.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO global_settings (name, value) VALUES ($1, $2)\n         ON CONFLICT (name) DO UPDATE SET value = $2, updated_at = now()",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "975e404ac3a6258bb8220e122e3de094c7ab23330fdbc74d6e4ad472ddd3c820"
+}

backend/.sqlx/query-9f663180166f53d117e794f3f3a5723a0a43db163ecca7d5a63d4e74ab1d3be1.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT pg_advisory_xact_lock(hashtext('datatable_role_catalog'))",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "pg_advisory_xact_lock",
+        "type_info": "Void"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "9f663180166f53d117e794f3f3a5723a0a43db163ecca7d5a63d4e74ab1d3be1"
+}

backend/.sqlx/query-a1b23f3e62c6433d95cdac58215741a51ca0ce66bf1c674097705cf2e1b72eff.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM ai_agent_memory a\n               USING flow_conversation c\n              WHERE c.id = ANY($1)\n                AND a.conversation_id = c.id\n                AND a.workspace_id = c.workspace_id\n                AND NOT EXISTS (\n                    SELECT 1 FROM flow_conversation_message m WHERE m.conversation_id = c.id\n                )",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "UuidArray"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "a1b23f3e62c6433d95cdac58215741a51ca0ce66bf1c674097705cf2e1b72eff"
+}

backend/.sqlx/query-a57de2bb0442a5ee8a607cd63cfcf675de175796184f620cb4b09670c8b0b19f.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture_config SET server_id = NULL, last_server_ping = NULL\n         WHERE workspace_id = $1 AND trigger_kind = 'postgres'\n           AND (trigger_config->>'postgres_resource_path' = $2\n                OR trigger_config->>'postgres_resource_path' LIKE $3)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "a57de2bb0442a5ee8a607cd63cfcf675de175796184f620cb4b09670c8b0b19f"
+}

backend/.sqlx/query-b42af37fb474bea4c5419b0a46d9eadfe384013ab970ccf9c5effd1c78321b7c.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT ws.datatable->'datatables' FROM workspace_settings ws\n             WHERE ws.workspace_id = $1 FOR UPDATE",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "b42af37fb474bea4c5419b0a46d9eadfe384013ab970ccf9c5effd1c78321b7c"
+}

backend/.sqlx/query-b9842d2d8abf382bd82d8fa1de012373638be391f884f81dc387ffc465badac6.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT jsonb_object_keys(value->'databases') FROM global_settings\n         WHERE name = 'custom_instance_pg_databases'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "jsonb_object_keys",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "b9842d2d8abf382bd82d8fa1de012373638be391f884f81dc387ffc465badac6"
+}

backend/.sqlx/query-bfdd60b42e32bd81e2d20b327462893147b4e5ff078531de36147d908132d636.json

@@ -1,14 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM flow_conversation_message WHERE job_id = ANY($1)",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "UuidArray"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "bfdd60b42e32bd81e2d20b327462893147b4e5ff078531de36147d908132d636"
-}

backend/.sqlx/query-c1d026c886799dabc39ce73e1fe09ccb175c7271df75d67aa9c72ad6f825a992.json

@@ -0,0 +1,28 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM usr WHERE email = $1 RETURNING username, workspace_id",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "username",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false
+    ]
+  },
+  "hash": "c1d026c886799dabc39ce73e1fe09ccb175c7271df75d67aa9c72ad6f825a992"
+}

backend/.sqlx/query-c1e3ed3ecc3bcb98f60ba8196d33fee4a74f61b061e5025ecb75882208b3ba8f.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "INSERT INTO flow_conversation (id, workspace_id, flow_path, created_by, title)\n         VALUES ($1, $2, $3, $4, $5)\n         RETURNING id, workspace_id, flow_path, title, created_at, updated_at, created_by",
+  "query": "INSERT INTO flow_conversation (id, workspace_id, flow_path, created_by, title)\n         VALUES ($1, $2, $3, $4, $5)\n         ON CONFLICT (id) DO NOTHING\n         RETURNING id, workspace_id, flow_path, title, created_at, updated_at, created_by",
   "describe": {
     "columns": [
       {
@@ -58,5 +58,5 @@
       false
     ]
   },
-  "hash": "6bd23a98838e3eec309e6b696edc776bd56fc9dae1238b3272557d1562400dbe"
+  "hash": "c1e3ed3ecc3bcb98f60ba8196d33fee4a74f61b061e5025ecb75882208b3ba8f"
 }

backend/.sqlx/query-c5451ea9d9fa5146af242d1ee8c19ebd65b80e7ed9f29b9fb2e03767c2aa94ba.json

@@ -0,0 +1,28 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT ws.workspace_id AS \"workspace_id!\", dt.key AS \"datatable!\"\n           FROM workspace_settings ws\n           CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt\n           WHERE dt.value->'reference'->>'workspace_id' = $1\n           ORDER BY ws.workspace_id, dt.key",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "datatable!",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      null
+    ]
+  },
+  "hash": "c5451ea9d9fa5146af242d1ee8c19ebd65b80e7ed9f29b9fb2e03767c2aa94ba"
+}

backend/.sqlx/query-c85d362fe2e652d4ac01a35bf470e80b993020a2ff5dcb5849dc570d52798587.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM datatable_role WHERE id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "c85d362fe2e652d4ac01a35bf470e80b993020a2ff5dcb5849dc570d52798587"
+}

backend/.sqlx/query-d48ca62c86b1af7a9dd2450c1c28dc45020a2a553d8874c49f9eafedea5a9d40.json

@@ -0,0 +1,24 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT dt.key AS \"datatable!\"\n           FROM workspace_settings ws\n           CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt\n           WHERE ws.workspace_id = $1\n             AND dt.key <> $2\n             AND NOT dt.value ? 'permissions'\n             AND dt.value->'database'->>'resource_type' = 'instance'\n             AND dt.value->'database'->>'resource_path' = $3\n           ORDER BY dt.key",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "datatable!",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "d48ca62c86b1af7a9dd2450c1c28dc45020a2a553d8874c49f9eafedea5a9d40"
+}

backend/.sqlx/query-d5fb5dde6300862f978739a3d9249fc2b3e7697c0da7d3195398933d3d81aadf.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT datatable FROM workspace_settings WHERE workspace_id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "datatable",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "d5fb5dde6300862f978739a3d9249fc2b3e7697c0da7d3195398933d3d81aadf"
+}

backend/.sqlx/query-da5057c86607327bafc2942c218025ca9181a0c396405984d87e422e129521c1.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace_settings ws\n               SET datatable = (\n                   SELECT jsonb_set(ws.datatable, '{datatables}', jsonb_object_agg(\n                       dt.key,\n                       CASE WHEN dt.value->'reference'->>'workspace_id' = $1\n                                 AND dt.value->'reference'->>'datatable' = $2\n                           THEN jsonb_set(dt.value, '{reference,datatable}', to_jsonb($3::text))\n                           ELSE dt.value END\n                   ))\n                   FROM jsonb_each(ws.datatable->'datatables') dt\n               )\n               WHERE jsonb_typeof(ws.datatable->'datatables') = 'object'\n                 AND ws.datatable::text LIKE '%\"reference\"%'",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "da5057c86607327bafc2942c218025ca9181a0c396405984d87e422e129521c1"
+}

backend/.sqlx/query-dc8dfc37559e9b6713bde48155f48b5a2c7b8199eace1508e102b60d1ff40c04.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT value->'roles' FROM global_settings WHERE name = 'custom_instance_pg_databases'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "dc8dfc37559e9b6713bde48155f48b5a2c7b8199eace1508e102b60d1ff40c04"
+}

backend/.sqlx/query-e159b2ff15633f85e839ee4fe1ec2ecd11caf228ea8d0f52ad66def595644250.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE postgres_trigger SET server_id = NULL, last_server_ping = NULL\n             WHERE workspace_id = $1\n               AND (postgres_resource_path = $2 OR postgres_resource_path LIKE $3)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "e159b2ff15633f85e839ee4fe1ec2ecd11caf228ea8d0f52ad66def595644250"
+}

backend/.sqlx/query-e2061df65ffd4a72146c4ca316829265289c8d6f625ac272655c88e1ad0b1745.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT datatable FROM workspace_settings WHERE workspace_id = $1 FOR UPDATE",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "datatable",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "e2061df65ffd4a72146c4ca316829265289c8d6f625ac272655c88e1ad0b1745"
+}

backend/.sqlx/query-ebaf3ed3097621da59dd201b5a4b9d1f440692f183c7c378f59e4b73f1c6e241.json

@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace_settings\n               SET datatable = jsonb_set(\n                   jsonb_set(\n                       datatable #- ARRAY['datatables', $2, 'reference'],\n                       ARRAY['datatables', $2, 'database'], $3::jsonb),\n                   ARRAY['datatables', $2, 'forked_from'], $4::jsonb\n               )\n               WHERE workspace_id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Jsonb",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "ebaf3ed3097621da59dd201b5a4b9d1f440692f183c7c378f59e4b73f1c6e241"
+}

backend/.sqlx/query-ec295b3890a0018475ec0a3774c7a30d71a5689efe72daf58bd1e8f6cf90c410.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id FROM flow_conversation WHERE id = ANY($1) ORDER BY id FOR UPDATE",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "UuidArray"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "ec295b3890a0018475ec0a3774c7a30d71a5689efe72daf58bd1e8f6cf90c410"
+}

backend/.sqlx/query-f3ee09fb17955ca8d886f446d397063c4094546a7807343b570b823796372cef.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT workspace_id FROM workspace_settings WHERE datatable::text LIKE $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "f3ee09fb17955ca8d886f446d397063c4094546a7807343b570b823796372cef"
+}

backend/.sqlx/query-f4adc9e26ebfebce18a29fb2c21bf06394cacb8a9699a608327b097e0ac1363e.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace_settings\n           SET datatable = CASE WHEN $3::jsonb = 'null'::jsonb\n               THEN datatable #- ARRAY['datatables', $2, 'permissions']\n               ELSE jsonb_set(datatable, ARRAY['datatables', $2, 'permissions'], $3::jsonb)\n           END\n           WHERE workspace_id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "f4adc9e26ebfebce18a29fb2c21bf06394cacb8a9699a608327b097e0ac1363e"
+}

backend/.sqlx/query-fcb34e643b888122766e115a01394ab31ac856252aaa76c75ea27a447009c363.json

@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE datatable_role SET name = $2, enabled = $3, pwd = $4 WHERE id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Varchar",
+        "Bool",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "fcb34e643b888122766e115a01394ab31ac856252aaa76c75ea27a447009c363"
+}

backend/Cargo.lock

@@ -15470,6 +15470,7 @@ dependencies = [
  "windmill-ai",
  "windmill-alerting",
  "windmill-api-auth",
+ "windmill-audit",
  "windmill-common",
  "windmill-object-store",
 ]
@@ -15652,6 +15653,7 @@ dependencies = [
  "pin-project-lite",
  "pkcs1",
  "postgres-native-tls 0.5.3",
+ "postgres-protocol",
  "prometheus",
  "quick_cache",
  "rand 0.9.0",

backend/Cargo.toml

@@ -624,6 +624,7 @@ wasm-bindgen-test = "^0"
 convert_case = "0.6.0"
 getrandom = "0.2"
 tokio-postgres = {version = "^0.7", features = ["array-impls", "with-serde_json-1", "with-chrono-0_4", "with-uuid-1", "with-bit-vec-0_6"]}
+postgres-protocol = "0.6"
 rust-postgres = { package = "tokio-postgres", git = "https://github.com/MaterializeInc/rust-postgres", rev = "78c1222577bb091d69bc22b1bc7ad01c14675abe"}
 rust-postgres-native-tls = { package = "postgres-native-tls", git = "https://github.com/MaterializeInc/rust-postgres", features = ["runtime"], rev = "78c1222577bb091d69bc22b1bc7ad01c14675abe" }
 bit-vec = "=0.6.3"

backend/ee-repo-ref.txt

@@ -1 +1 @@
-d252afcc80e77fcc4f9a2a346b80908c8605a6c0
+2de95863062cc0b00933afd86800aaf953af47d5

backend/migrations/20260908142148_datatable_role_catalog.down.sql

@@ -0,0 +1,18 @@
+-- Refuse while the catalog holds anything. Each row is a live Postgres login with a password
+-- only this table carries, so dropping it would leave credentials on the cluster that Windmill can
+-- no longer disable, delete or even name — and re-applying could not recreate them, because the
+-- role names would already be taken. Cleaning them up here is not an option either: dropping a
+-- role means reassigning what it owns in *every* instance database, and a migration runs in one.
+--
+-- Delete the roles through instance settings first; that path does the cluster work.
+LOCK TABLE datatable_role IN ACCESS EXCLUSIVE MODE;
+
+DO $$
+BEGIN
+    IF EXISTS (SELECT 1 FROM datatable_role) THEN
+        RAISE EXCEPTION 'Cannot roll back: % data table role(s) still exist as Postgres logins. Delete them in instance settings first, which drops them from the cluster.',
+            (SELECT count(*) FROM datatable_role);
+    END IF;
+END $$;
+
+DROP TABLE IF EXISTS datatable_role;

backend/migrations/20260908142148_datatable_role_catalog.up.sql

@@ -0,0 +1,21 @@
+-- The instance's data table role catalog: one row per Postgres login Windmill created for data
+-- table access.
+--
+-- A table rather than a `global_settings` key, because the value is a set of live cluster
+-- credentials and that table has generic read, list, write and CLI round-trip paths that know
+-- nothing about what they are carrying. Every one of them is a way to leak the passwords or to
+-- overwrite the catalog with a copy that has none, and a row nothing generic touches has none of
+-- those. One row per role also makes two concurrent creates two inserts rather than a
+-- read-modify-write over one document.
+CREATE TABLE datatable_role (
+    id VARCHAR(50) PRIMARY KEY,
+    -- The Postgres role name, verbatim. Unique because it is the cluster's own key.
+    name VARCHAR(63) NOT NULL UNIQUE,
+    enabled BOOLEAN NOT NULL DEFAULT true,
+    -- Generated by Windmill, never entered by anyone, and never leaves the server.
+    pwd TEXT,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+GRANT ALL ON datatable_role TO windmill_user;
+GRANT ALL ON datatable_role TO windmill_admin;

backend/parsers/windmill-parser/src/asset_parser.rs

@@ -730,7 +730,12 @@ pub fn parse_asset_syntax(
     s: &str,
     enable_default_syntax: bool,
 ) -> Option<(AssetKind, Cow<'_, str>)> {
-    if enable_default_syntax && s == "datatable" {
+    // `datatable` and `datatable?role=analyst` both name the default data table: the role picks
+    // which Postgres login the connection is made as, not which data table is read.
+    if enable_default_syntax
+        && s.strip_prefix("datatable")
+            .is_some_and(|rest| rest.is_empty() || rest.starts_with('?'))
+    {
         return Some((AssetKind::DataTable, Cow::Borrowed("main")));
     } else if enable_default_syntax && s == "ducklake" {
         return Some((AssetKind::Ducklake, Cow::Borrowed("main")));
@@ -741,6 +746,14 @@ pub fn parse_asset_syntax(
             if *kind == AssetKind::Dbt {
                 return Some((*kind, Cow::Owned(canonicalize_table_asset_path(suffix))));
             }
+            // Same reasoning as above, for the explicit form. Specific to data tables: a
+            // `Resource`'s `?table=` is part of what it names, and stripping it would merge two
+            // different assets.
+            if *kind == AssetKind::DataTable {
+                if let Some((path, _role)) = suffix.split_once('?') {
+                    return Some((*kind, Cow::Borrowed(path)));
+                }
+            }
             // The suffix is kept verbatim. For S3 the path encodes the storage:
             // `s3://<storage>/<key>`, with an EMPTY storage segment for the
             // workspace default — so `s3:///key` yields `/key` (leading slash
@@ -1692,6 +1705,25 @@ fn parse_trigger_spec(s: &str) -> Option<TriggerSpec> {
 mod pipeline_annotation_tests {
     use super::*;
 
+    #[test]
+    fn a_datatable_role_is_not_part_of_the_asset_it_names() {
+        // The role picks which Postgres login the connection is made as, so two references that
+        // differ only by role are the same asset and must land on one graph node.
+        assert_eq!(
+            parse_asset_syntax("datatable://sales?role=analytics", false),
+            Some((AssetKind::DataTable, Cow::Borrowed("sales")))
+        );
+        assert_eq!(
+            parse_asset_syntax("datatable?role=analytics", true),
+            Some((AssetKind::DataTable, Cow::Borrowed("main")))
+        );
+        // A resource's `?table=` is part of what it names, so it is kept.
+        assert_eq!(
+            parse_asset_syntax("$res:f/db/pg?table=users", false),
+            Some((AssetKind::Resource, Cow::Borrowed("f/db/pg?table=users")))
+        );
+    }
+
     #[test]
     fn s3_path_keeps_storage_distinction() {
         // An S3 asset path is `<storage>/<key>` with an empty storage segment

backend/tests/v2_job_delete_orphans.rs

@@ -37,7 +37,7 @@ async fn seed_side_rows(db: &Pool<Postgres>, ws: &str, job_id: Uuid) -> anyhow::
     .bind(ws)
     .execute(db)
     .await?;
-    // created_seq is assigned by a trigger; inserting a value is rejected.
+    // created_seq is an identity column; supplying a value is rejected.
     sqlx::query(
         "INSERT INTO flow_conversation_message (conversation_id, message_type, content, job_id)
          VALUES ($1, 'assistant', 'hi', $2)",
@@ -121,6 +121,184 @@ async fn test_delete_jobs_removes_side_rows(db: Pool<Postgres>) -> anyhow::Resul
     Ok(())
 }
 
+/// (conversation rows, agent-memory rows) for one conversation.
+async fn conversation_and_memory_counts(
+    db: &Pool<Postgres>,
+    conversation_id: Uuid,
+) -> anyhow::Result<(i64, i64)> {
+    Ok((
+        count(
+            db,
+            "SELECT count(*) FROM flow_conversation WHERE id = $1",
+            conversation_id,
+        )
+        .await?,
+        count(
+            db,
+            "SELECT count(*) FROM ai_agent_memory WHERE conversation_id = $1",
+            conversation_id,
+        )
+        .await?,
+    ))
+}
+
+/// A conversation outlives the jobs behind its messages until the last one goes: only then
+/// are the row and the agent's memory for it left with nothing, and only then are they
+/// deleted. Both halves matter — the surviving half is what a single data-modifying CTE
+/// would break, since its emptiness check would read the snapshot from before the delete.
+#[sqlx::test(fixtures("base"))]
+async fn test_delete_jobs_removes_a_conversation_once_its_last_message_goes(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    let first_job = Uuid::new_v4();
+    let second_job = Uuid::new_v4();
+    insert_job(&db, WS, first_job).await?;
+    insert_job(&db, WS, second_job).await?;
+
+    let conv_id = Uuid::new_v4();
+    sqlx::query(
+        "INSERT INTO flow_conversation (id, workspace_id, flow_path, created_by)
+         VALUES ($1, $2, 'f/flow', 'test-user')",
+    )
+    .bind(conv_id)
+    .bind(WS)
+    .execute(&db)
+    .await?;
+    for job_id in [first_job, second_job] {
+        sqlx::query(
+            "INSERT INTO flow_conversation_message (conversation_id, message_type, content, job_id)
+             VALUES ($1, 'assistant', 'hi', $2)",
+        )
+        .bind(conv_id)
+        .bind(job_id)
+        .execute(&db)
+        .await?;
+    }
+    sqlx::query(
+        "INSERT INTO ai_agent_memory (workspace_id, conversation_id, step_id, messages)
+         VALUES ($1, $2, 'a', '[]'::jsonb)",
+    )
+    .bind(WS)
+    .bind(conv_id)
+    .execute(&db)
+    .await?;
+
+    let mut conn = db.acquire().await?;
+    windmill_common::jobs::delete_jobs(&mut conn, &[first_job]).await?;
+    drop(conn);
+    assert_eq!(
+        conversation_and_memory_counts(&db, conv_id).await?,
+        (1, 1),
+        "a conversation with a message left must survive, memory included"
+    );
+
+    let mut conn = db.acquire().await?;
+    windmill_common::jobs::delete_jobs(&mut conn, &[second_job]).await?;
+    drop(conn);
+    assert_eq!(
+        conversation_and_memory_counts(&db, conv_id).await?,
+        (0, 0),
+        "the last message going should take the conversation and its memory"
+    );
+    Ok(())
+}
+
+/// Turns that start while retention is collecting their conversation must land, not fail:
+/// the conversation lookup locks the row, so each turn waits for the collector's commit,
+/// finds the conversation gone, and creates it again — the first insert wins and the other
+/// reads its row. Without the lock a turn's message insert is what waits, on the parent
+/// row's key lock, and fails its FK check afterwards.
+#[sqlx::test(fixtures("base"))]
+async fn test_new_turns_wait_for_conversation_cleanup_and_recreate(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    let old_job = Uuid::new_v4();
+    let new_jobs = [Uuid::new_v4(), Uuid::new_v4()];
+    insert_job(&db, WS, old_job).await?;
+    for job in new_jobs {
+        insert_job(&db, WS, job).await?;
+    }
+
+    let conv_id = Uuid::new_v4();
+    sqlx::query(
+        "INSERT INTO flow_conversation (id, workspace_id, flow_path, created_by)
+         VALUES ($1, $2, 'f/flow', 'test-user')",
+    )
+    .bind(conv_id)
+    .bind(WS)
+    .execute(&db)
+    .await?;
+    sqlx::query(
+        "INSERT INTO flow_conversation_message (conversation_id, message_type, content, job_id)
+         VALUES ($1, 'user', 'hi', $2)",
+    )
+    .bind(conv_id)
+    .bind(old_job)
+    .execute(&db)
+    .await?;
+
+    // The collector holds the conversation row locked and deleted, uncommitted.
+    let mut cleanup = db.begin().await?;
+    windmill_common::jobs::delete_jobs(&mut *cleanup, &[old_job]).await?;
+
+    let turns: Vec<_> = new_jobs
+        .into_iter()
+        .map(|new_job| {
+            let db = db.clone();
+            tokio::spawn(async move {
+                let mut tx = db.begin().await?;
+                windmill_common::flow_conversations::get_or_create_conversation_with_id(
+                    &mut tx,
+                    WS,
+                    "f/flow",
+                    "test-user",
+                    "hi again",
+                    conv_id,
+                )
+                .await?;
+                windmill_common::flow_conversations::add_message_to_conversation_tx(
+                    &mut tx,
+                    conv_id,
+                    Some(new_job),
+                    "hi again",
+                    windmill_common::flow_conversations::MessageType::User,
+                    None,
+                    true,
+                )
+                .await?;
+                tx.commit().await?;
+                anyhow::Ok(())
+            })
+        })
+        .collect();
+    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
+    cleanup.commit().await?;
+    for turn in turns {
+        turn.await??;
+    }
+
+    assert_eq!(
+        conversation_and_memory_counts(&db, conv_id).await?.0,
+        1,
+        "the turns must have created the conversation again, once"
+    );
+    assert_eq!(
+        count(
+            &db,
+            "SELECT count(*) FROM flow_conversation_message WHERE conversation_id = $1",
+            conv_id,
+        )
+        .await?,
+        2,
+        "both turns' messages should be there"
+    );
+    Ok(())
+}
+
 #[sqlx::test(fixtures("base"))]
 async fn test_clear_schedule_removes_side_rows(db: Pool<Postgres>) -> anyhow::Result<()> {
     initialize_tracing().await;
@@ -192,6 +370,74 @@ async fn test_workspace_delete_removes_side_rows(db: Pool<Postgres>) -> anyhow::
     Ok(())
 }
 
+/// The purge endpoint carries its own copy of the emptied-conversation rule, so it gets the
+/// same guard: the conversation and its memory go with the last message, and not before.
+#[sqlx::test(fixtures("base"))]
+async fn test_jobs_export_delete_removes_a_conversation_once_its_last_message_goes(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    let first_job = Uuid::new_v4();
+    let second_job = Uuid::new_v4();
+    insert_job(&db, WS, first_job).await?;
+    insert_job(&db, WS, second_job).await?;
+
+    let conv_id = Uuid::new_v4();
+    sqlx::query(
+        "INSERT INTO flow_conversation (id, workspace_id, flow_path, created_by)
+         VALUES ($1, $2, 'f/flow', 'test-user')",
+    )
+    .bind(conv_id)
+    .bind(WS)
+    .execute(&db)
+    .await?;
+    for job_id in [first_job, second_job] {
+        sqlx::query(
+            "INSERT INTO flow_conversation_message (conversation_id, message_type, content, job_id)
+             VALUES ($1, 'assistant', 'hi', $2)",
+        )
+        .bind(conv_id)
+        .bind(job_id)
+        .execute(&db)
+        .await?;
+    }
+    sqlx::query(
+        "INSERT INTO ai_agent_memory (workspace_id, conversation_id, step_id, messages)
+         VALUES ($1, $2, 'a', '[]'::jsonb)",
+    )
+    .bind(WS)
+    .bind(conv_id)
+    .execute(&db)
+    .await?;
+
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    let purge = |job_id: Uuid| async move {
+        reqwest::Client::new()
+            .post(format!("http://localhost:{port}/api/w/{WS}/jobs/delete"))
+            .header("Authorization", "Bearer SECRET_TOKEN")
+            .json(&[job_id])
+            .send()
+            .await
+    };
+
+    assert!(purge(first_job).await?.status().is_success());
+    assert_eq!(
+        conversation_and_memory_counts(&db, conv_id).await?,
+        (1, 1),
+        "a conversation with a message left must survive the purge endpoint too"
+    );
+
+    assert!(purge(second_job).await?.status().is_success());
+    assert_eq!(
+        conversation_and_memory_counts(&db, conv_id).await?,
+        (0, 0),
+        "the last message going should take the conversation and its memory"
+    );
+    Ok(())
+}
+
 /// The `/jobs/delete` purge endpoint must scope every side-table delete to the path
 /// workspace. A `test-workspace` admin passing a job id from another workspace must not be
 /// able to delete that workspace's job or side rows (the side tables no longer cascade, so

backend/windmill-api-groups/src/folders.rs

@@ -800,6 +800,14 @@ async fn delete_folder(
 
     not_found_if_none(get_folderopt(&mut tx, &w_id, &name).await?, "Folder", &name)?;
 
+    // See the same call in `delete_group`: a freed name must not stay in a tenant list.
+    windmill_common::workspaces::remove_datatable_tenant_in_workspace(
+        &mut tx,
+        &w_id,
+        &format!("f/{name}"),
+    )
+    .await?;
+
     let del = sqlx::query_scalar!(
         "DELETE FROM folder WHERE name = $1 AND workspace_id = $2 RETURNING 1",
         name,

backend/windmill-api-groups/src/groups.rs

@@ -797,6 +797,15 @@ async fn delete_group(
     }
     not_found_if_none(get_group_opt(&mut tx, &w_id, &name).await?, "Group", &name)?;
 
+    // A tenant list names a principal, so a freed name must not linger in one: a later group
+    // reusing it would silently inherit the data table access this one had.
+    windmill_common::workspaces::remove_datatable_tenant_in_workspace(
+        &mut tx,
+        &w_id,
+        &format!("g/{name}"),
+    )
+    .await?;
+
     sqlx::query!(
         "DELETE FROM usr_to_group WHERE group_ = $1 AND workspace_id = $2",
         name,

backend/windmill-api-integration-tests/tests/datatable_acl.rs

@@ -0,0 +1,108 @@
+//! Who may read and change a data table's grants and owners. On the Enterprise Edition: its
+//! administrators, from the workspace that governs it. Without it: nobody. Each refusal is decided
+//! before anything connects to the data table, so the fixture's database never has to exist.
+
+use serde_json::{json, Value};
+use sqlx::{Pool, Postgres};
+
+use windmill_test_utils::*;
+
+fn grant_select_on_public() -> Value {
+    json!({
+        "target": {"kind": "schema", "schema": "public"},
+        "change": {"type": "grant", "role": "analytics", "privileges": ["SELECT"],
+                   "scope": "all_tables"},
+        "statements": [r#"GRANT SELECT ON ALL TABLES IN SCHEMA "public" TO "analytics""#]
+    })
+}
+
+async fn post_acl(
+    port: u16,
+    w_id: &str,
+    action: &str,
+    token: &str,
+) -> anyhow::Result<reqwest::Response> {
+    Ok(reqwest::Client::new()
+        .post(format!(
+            "http://localhost:{port}/api/w/{w_id}/workspaces/datatable_acl/main/{action}"
+        ))
+        .header("Authorization", format!("Bearer {token}"))
+        .json(&grant_select_on_public())
+        .send()
+        .await?)
+}
+
+/// A fork reaches the data table through a pointer: it may use it, never change what each role may
+/// touch on it — not even as an admin of the fork.
+#[cfg(all(feature = "private", feature = "enterprise"))]
+#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
+async fn a_fork_cannot_change_access_on_the_data_table_it_points_at(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    for action in ["plan", "apply"] {
+        let resp = post_acl(port, "wm-fork-dt", action, "SECRET_TOKEN_2").await?;
+        assert_eq!(resp.status(), 401, "{action}: {}", resp.text().await?);
+    }
+    Ok(())
+}
+
+#[cfg(all(feature = "private", feature = "enterprise"))]
+#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
+async fn a_member_who_is_not_an_admin_cannot_change_access(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    for action in ["plan", "apply"] {
+        let resp = post_acl(port, "test-workspace", action, "SECRET_TOKEN_2").await?;
+        assert_eq!(resp.status(), 401, "{action}: {}", resp.text().await?);
+    }
+    Ok(())
+}
+
+/// Not even reading, and not even on a data table that is not under roles — which any member
+/// reaches, so only the edition stands between them and the instance's credentials.
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
+async fn only_the_enterprise_edition_has_the_access_editor(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    sqlx::query(
+        "UPDATE workspace_settings
+         SET datatable = datatable #- '{datatables,main,permissions}'
+         WHERE workspace_id = 'test-workspace'",
+    )
+    .execute(&db)
+    .await?;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+
+    let read = reqwest::Client::new()
+        .get(format!(
+            "http://localhost:{port}/api/w/test-workspace/workspaces/datatable_acl/main?kind=database"
+        ))
+        .header("Authorization", "Bearer SECRET_TOKEN")
+        .send()
+        .await?;
+    let mut responses = vec![("read", read)];
+    for action in ["plan", "apply"] {
+        responses.push((
+            action,
+            post_acl(port, "test-workspace", action, "SECRET_TOKEN").await?,
+        ));
+    }
+    for (action, resp) in responses {
+        assert_eq!(resp.status(), 400, "{action}");
+        let body = resp.text().await?;
+        assert!(
+            body.contains("Data table roles are a Windmill Enterprise Edition feature"),
+            "{action}: {body}"
+        );
+    }
+    Ok(())
+}

backend/windmill-api-integration-tests/tests/fixtures/datatable_roles.sql

@@ -0,0 +1,45 @@
+-- A data table under roles in `test-workspace`, and a fork whose entry points at it rather than
+-- carrying a copy. `test-user-2` is a non-admin of the parent and an admin of the fork: the shape
+-- the pointer exists for.
+
+-- Empty registry: role provisioning grants CONNECT on every database named here, and the data
+-- table's `dt_main` is a name in workspace settings, not a database that exists.
+INSERT INTO global_settings (name, value) VALUES
+    ('custom_instance_pg_databases', '{"user_pwd": "pw", "databases": {}}'::jsonb)
+    ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value;
+
+INSERT INTO datatable_role (id, name, enabled, pwd) VALUES ('role1', 'analytics', true, 'pw');
+
+UPDATE workspace_settings SET datatable = '{
+    "datatables": {
+        "main": {
+            "database": {"resource_type": "instance", "resource_path": "dt_main"},
+            "permissions": {
+                "default_role": "role1",
+                "roles": {
+                    "admin": {"tenants": []},
+                    "role1": {"tenants": ["u/test-user-2", "g/analysts", "f/finance"]}
+                }
+            }
+        }
+    }
+}'::jsonb WHERE workspace_id = 'test-workspace';
+
+INSERT INTO group_ (workspace_id, name, summary, extra_perms) VALUES
+	('test-workspace', 'analysts', 'Analysts', '{}');
+INSERT INTO folder (workspace_id, name, display_name, owners, extra_perms) VALUES
+	('test-workspace', 'finance', 'finance', '{}', '{}');
+
+INSERT INTO workspace (id, name, owner, parent_workspace_id) VALUES
+	('wm-fork-dt', 'fork of test-workspace', 'test2@windmill.dev', 'test-workspace');
+INSERT INTO workspace_key (workspace_id, kind, key) VALUES ('wm-fork-dt', 'cloud', 'test-key');
+INSERT INTO group_ (workspace_id, name, summary, extra_perms) VALUES
+	('wm-fork-dt', 'all', 'All users', '{}');
+INSERT INTO usr (workspace_id, email, username, is_admin, role) VALUES
+	('wm-fork-dt', 'test2@windmill.dev', 'test-user-2', true, 'Admin');
+
+INSERT INTO workspace_settings (workspace_id, datatable) VALUES ('wm-fork-dt', '{
+    "datatables": {
+        "main": {"reference": {"workspace_id": "test-workspace", "datatable": "main"}}
+    }
+}'::jsonb);

backend/windmill-api-jobs/src/execution.rs

@@ -668,10 +668,16 @@ pub async fn handle_chat_conversation_messages(
     flow_path: &str,
     run_query: &RunJobQuery,
     user_message_raw: Option<&Box<serde_json::value::RawValue>>,
+    job_id: Uuid,
 ) -> error::Result<()> {
+    // Names the query parameter rather than the field: it is not a flow argument, and
+    // supplying it as one is the first thing tried on reading `memory_id is required`.
     let memory_id = run_query.memory_id.ok_or_else(|| {
         windmill_common::error::Error::BadRequest(
-            "memory_id is required for chat-enabled flows".to_string(),
+            "memory_id is required for chat-enabled flows. Pass it as the `memory_id` query \
+             parameter, not as a flow argument: it names the conversation the turn belongs to, \
+             so a fresh UUID starts one and reusing a UUID continues it."
+                .to_string(),
         )
     })?;
 
@@ -698,10 +704,13 @@ pub async fn handle_chat_conversation_messages(
     )
     .await?;
 
+    // The run this message started. Its args are the only record of what the message
+    // carried besides its text — attachments and every other flow input — and nothing
+    // written later points at them: an assistant row holds the AI agent step's job.
     add_message_to_conversation_tx(
         tx,
         memory_id,
-        None,
+        Some(job_id),
         &user_message,
         MessageType::User,
         None,
@@ -826,6 +835,7 @@ pub async fn run_flow<'c>(
             &flow_path.to_string(),
             &run_query,
             args.args.get("user_message"),
+            uuid,
         )
         .await?;
     }

backend/windmill-api-jobs/src/jobs_export.rs

@@ -692,16 +692,64 @@ pub async fn delete_jobs(
     .await?
     .rows_affected();
 
-    let conversation_message_deleted = sqlx::query!(
+    // One row per message deleted, so the conversation of a chat losing several appears
+    // several times: the count is taken before the dedup below.
+    let mut conversation_ids: Vec<Uuid> = sqlx::query_scalar!(
         "DELETE FROM flow_conversation_message m
          USING flow_conversation c
-         WHERE m.conversation_id = c.id AND c.workspace_id = $1 AND m.job_id = ANY($2)",
+         WHERE m.conversation_id = c.id AND c.workspace_id = $1 AND m.job_id = ANY($2)
+         RETURNING m.conversation_id",
         &w_id,
         &job_ids
     )
-    .execute(&mut *tx)
-    .await?
-    .rows_affected();
+    .fetch_all(&mut *tx)
+    .await?;
+    let conversation_message_deleted = conversation_ids.len() as u64;
+
+    // Same rule, lock and statement order as retention (windmill_common::jobs::delete_jobs,
+    // which says why): a conversation with no messages left goes, and its memory with it.
+    conversation_ids.sort_unstable();
+    conversation_ids.dedup();
+    let mut memory_deleted = 0;
+    let mut conversation_deleted = 0;
+    if !conversation_ids.is_empty() {
+        sqlx::query_scalar!(
+            "SELECT id FROM flow_conversation WHERE id = ANY($1) AND workspace_id = $2 ORDER BY id FOR UPDATE",
+            &conversation_ids,
+            &w_id
+        )
+        .fetch_all(&mut *tx)
+        .await?;
+        memory_deleted = sqlx::query!(
+            "DELETE FROM ai_agent_memory a
+               USING flow_conversation c
+              WHERE c.id = ANY($1)
+                AND c.workspace_id = $2
+                AND a.conversation_id = c.id
+                AND a.workspace_id = c.workspace_id
+                AND NOT EXISTS (
+                    SELECT 1 FROM flow_conversation_message m WHERE m.conversation_id = c.id
+                )",
+            &conversation_ids,
+            &w_id
+        )
+        .execute(&mut *tx)
+        .await?
+        .rows_affected();
+        conversation_deleted = sqlx::query!(
+            "DELETE FROM flow_conversation c
+              WHERE c.id = ANY($1)
+                AND c.workspace_id = $2
+                AND NOT EXISTS (
+                    SELECT 1 FROM flow_conversation_message m WHERE m.conversation_id = c.id
+                )",
+            &conversation_ids,
+            &w_id
+        )
+        .execute(&mut *tx)
+        .await?
+        .rows_affected();
+    }
 
     // Resolutions are not exported, so a delete-then-reimport of the same UUID would
     // otherwise resurrect the old annotation on a job that never carried one.
@@ -737,6 +785,8 @@ pub async fn delete_jobs(
         + zombie_deleted
         + dispatch_event_deleted
         + conversation_message_deleted
+        + memory_deleted
+        + conversation_deleted
         + resolution_deleted
         + jobs_deleted;
 

backend/windmill-api-settings/Cargo.toml

@@ -11,7 +11,7 @@ path = "src/lib.rs"
 [features]
 default = []
 enterprise = ["license"]
-private = ["windmill-common/private"]
+private = ["windmill-common/private", "windmill-audit/private"]
 parquet = ["windmill-common/parquet", "windmill-object-store/parquet"]
 license = ["dep:rsa"]
 
@@ -19,6 +19,7 @@ license = ["dep:rsa"]
 windmill-ai = { workspace = true, default-features = false }
 windmill-alerting.workspace = true
 windmill-api-auth.workspace = true
+windmill-audit.workspace = true
 windmill-common = { workspace = true, default-features = false }
 axum.workspace = true
 anyhow.workspace = true

backend/windmill-api-settings/src/datatable_roles_oss.rs

@@ -0,0 +1,44 @@
+/*
+ * Author: Ruben Fiszel
+ * Copyright: Windmill Labs, Inc 2022
+ * This file and its contents are licensed under the AGPLv3 License.
+ * Please see the included NOTICE for copyright information and
+ * LICENSE-AGPL for a copy of the license.
+ */
+
+//! Where the data table role catalog endpoints come from: the enterprise implementation, or a
+//! refusal. Roles are an Enterprise Edition feature; see `windmill_common::datatable_roles_oss`.
+
+#[cfg(all(feature = "private", feature = "enterprise"))]
+pub(crate) use crate::datatable_roles_ee::{
+    create_datatable_role, delete_datatable_role, list_datatable_roles, update_datatable_role,
+};
+
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+pub(crate) use ce::*;
+
+// The routes stay registered so the API has one shape; each answers after authentication, before
+// anything is read.
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+mod ce {
+    use windmill_api_auth::ApiAuthed;
+    use windmill_common::{
+        datatable_roles_oss::datatable_roles_unavailable as unavailable, error::Result,
+    };
+
+    pub(crate) async fn list_datatable_roles(_authed: ApiAuthed) -> Result<String> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn create_datatable_role(_authed: ApiAuthed) -> Result<String> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn update_datatable_role(_authed: ApiAuthed) -> Result<String> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn delete_datatable_role(_authed: ApiAuthed) -> Result<String> {
+        Err(unavailable())
+    }
+}

backend/windmill-api-settings/src/lib.rs

@@ -17,6 +17,9 @@ mod audit_logs_s3;
 mod audit_logs_s3_backfill;
 #[cfg(feature = "parquet")]
 mod background_task;
+#[cfg(all(feature = "private", feature = "enterprise"))]
+mod datatable_roles_ee;
+mod datatable_roles_oss;
 #[cfg(feature = "private")]
 mod ee;
 pub mod ee_oss;
@@ -57,7 +60,7 @@ use windmill_common::{
     global_settings::{
         AI_CONFIG_SETTING, APP_WORKSPACED_ROUTE_SETTING, AUTOMATE_USERNAME_CREATION_SETTING,
         CRITICAL_ALERT_MUTE_UI_SETTING, CUSTOM_TAGS_SETTING, DEFAULT_TAGS_WORKSPACES_SETTING,
-        DISABLE_HUB_SETTING, EMAIL_DOMAIN_SETTING, ENV_SETTINGS,
+        DISABLE_HUB_SETTING, EMAIL_DOMAIN_SETTING, ENV_SETTINGS, EXTERNAL_INSTANCE_PG_SETTING,
         GITHUB_APP_WEBHOOK_BASE_URL_SETTING, HTTP_ROUTE_DEFAULT_ALLOWED_ORIGINS_SETTING,
         HTTP_ROUTE_WORKSPACED_ROUTE_SETTING, HUB_ACCESSIBLE_URL_SETTING, HUB_BASE_URL_SETTING,
         INSTANCE_BANNER_SETTING, MAX_RETENTION_OVERRIDE_WORKSPACES,
@@ -151,10 +154,28 @@ pub fn global_service() -> Router {
             "/list_custom_instance_pg_databases",
             post(list_custom_instance_pg_databases),
         )
+        .route(
+            "/datatable_roles",
+            get(datatable_roles_oss::list_datatable_roles)
+                .post(datatable_roles_oss::create_datatable_role),
+        )
+        .route(
+            "/datatable_roles/{id}",
+            post(datatable_roles_oss::update_datatable_role)
+                .delete(datatable_roles_oss::delete_datatable_role),
+        )
         .route(
             "/refresh_custom_instance_user_pwd",
             post(refresh_custom_instance_user_pwd),
         )
+        .route(
+            "/external_instance_pg/status",
+            get(get_external_instance_pg_status),
+        )
+        .route(
+            "/external_instance_pg/setup",
+            post(setup_external_instance_pg),
+        )
         .route(
             "/setup_custom_instance_pg_database/{name}",
             post(setup_custom_instance_pg_database),
@@ -925,6 +946,13 @@ async fn run_setting_pre_write_hook(
     value: &serde_json::Value,
 ) -> error::Result<()> {
     match key {
+        EXTERNAL_INSTANCE_PG_SETTING => {
+            windmill_common::external_instance_pg::check_external_instance_pg_write(
+                db,
+                Some(value),
+            )
+            .await?;
+        }
         // The instance AI config is written as an untyped blob through this generic
         // endpoint, so it never passes the typed check the workspace handler applies.
         // Rates that reach a cost total unbounded would make it negative or infinite.
@@ -1276,6 +1304,14 @@ async fn set_instance_config(
         for (key, value) in &settings_diff.upserts {
             run_setting_pre_write_hook(&db, key, value).await?;
         }
+        if settings_diff
+            .deletes
+            .iter()
+            .any(|k| k == EXTERNAL_INSTANCE_PG_SETTING)
+        {
+            windmill_common::external_instance_pg::check_external_instance_pg_write(&db, None)
+                .await?;
+        }
 
         instance_config::apply_settings_diff(&db, &settings_diff)
             .await
@@ -1730,6 +1766,54 @@ async fn refresh_custom_instance_user_pwd(
     Ok(Json(()))
 }
 
+async fn get_external_instance_pg_status(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+) -> JsonResult<windmill_common::external_instance_pg::ExternalInstancePgStatus> {
+    require_super_admin(&db, &authed).await?;
+    Ok(Json(
+        windmill_common::external_instance_pg::external_instance_pg_status(&db).await?,
+    ))
+}
+
+#[derive(Deserialize)]
+struct SetupExternalInstancePgBody {
+    #[serde(default)]
+    rotate_passwords: bool,
+}
+
+async fn setup_external_instance_pg(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Json(body): Json<SetupExternalInstancePgBody>,
+) -> JsonResult<windmill_common::external_instance_pg::ExternalInstancePgSetupReport> {
+    require_super_admin(&db, &authed).await?;
+    let report = windmill_common::external_instance_pg::setup_external_instance_pg_unchecked(
+        &db,
+        body.rotate_passwords,
+    )
+    .await?;
+    let rotated = body.rotate_passwords.to_string();
+    let success = report.success.to_string();
+    windmill_audit::audit_oss::audit_log(
+        &db,
+        &authed,
+        "settings.setup_external_instance_pg",
+        windmill_audit::ActionKind::Update,
+        "global",
+        Some(&authed.email),
+        Some(
+            [
+                ("rotate_passwords", rotated.as_str()),
+                ("success", success.as_str()),
+            ]
+            .into(),
+        ),
+    )
+    .await?;
+    Ok(Json(report))
+}
+
 #[derive(Deserialize)]
 struct SetupCustomInstanceDbBody {
     tag: Option<String>,

backend/windmill-api-users/src/users.rs

@@ -1703,14 +1703,25 @@ async fn delete_user(
         .await?;
     windmill_common::user_drafts::delete_drafts_of_email(&mut *tx, &email_to_delete).await?;
 
-    let usernames = sqlx::query_scalar!(
-        "DELETE FROM usr WHERE email = $1 RETURNING username",
+    let memberships = sqlx::query!(
+        "DELETE FROM usr WHERE email = $1 RETURNING username, workspace_id",
         &email_to_delete
     )
     .fetch_all(&mut *tx)
     .await?;
 
-    for username in usernames {
+    for row in memberships {
+        let username = row.username;
+        // A tenant list names a principal of its workspace, so the name has to be freed in every
+        // workspace this account belonged to: a later account taking the username would otherwise
+        // inherit the data table access it had.
+        windmill_common::workspaces::remove_datatable_tenant_in_workspace(
+            &mut tx,
+            &row.workspace_id,
+            &format!("u/{username}"),
+        )
+        .await?;
+
         sqlx::query!("DELETE FROM password WHERE email = $1", &email_to_delete)
             .execute(&mut *tx)
             .await?;
@@ -2456,6 +2467,15 @@ pub async fn delete_workspace_user_internal(
     tx: &mut Transaction<'_, Postgres>,
     authed: Option<&ApiAuthed>, // None for system operations
 ) -> Result<()> {
+    // Same reasoning as the `extra_perms` sweep below: a freed username must not stay named
+    // anywhere that grants access, tenant lists included.
+    windmill_common::workspaces::remove_datatable_tenant_in_workspace(
+        tx,
+        w_id,
+        &format!("u/{username_to_delete}"),
+    )
+    .await?;
+
     // ---- Clean up extra_perms referencing this user ----
     let extra_perms_tables = [
         "script",
@@ -3965,6 +3985,12 @@ async fn leave_workspace(
 ) -> Result<String> {
     forbid_job_token_account_destruction(&authed)?;
     let mut tx = db.begin().await?;
+    windmill_common::workspaces::remove_datatable_tenant_in_workspace(
+        &mut tx,
+        &w_id,
+        &format!("u/{}", authed.username),
+    )
+    .await?;
     sqlx::query!(
         "DELETE FROM usr WHERE workspace_id = $1 AND username = $2",
         &w_id,

backend/windmill-api-workspaces/src/datatable_acl_oss.rs

@@ -0,0 +1,44 @@
+/*
+ * Author: Ruben Fiszel
+ * Copyright: Windmill Labs, Inc 2022
+ * This file and its contents are licensed under the AGPLv3 License.
+ * Please see the included NOTICE for copyright information and
+ * LICENSE-AGPL for a copy of the license.
+ */
+
+//! Where the ACL planner comes from: the enterprise one, or a refusal.
+//!
+//! Data table roles are an Enterprise Edition feature, and so is everything here — reading who
+//! owns what included. `private` alone is not that edition — community builds carry it — so the
+//! planner is behind `enterprise` as well.
+
+#[cfg(all(feature = "private", feature = "enterprise"))]
+pub(crate) use crate::datatable_acl_ee::plan_statements;
+
+#[cfg(all(feature = "private", feature = "enterprise"))]
+pub(crate) fn ensure_datatable_acl_available() -> windmill_common::error::Result<()> {
+    Ok(())
+}
+
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+use {
+    crate::datatable_acl::{AclChange, AclPlan, AclTarget, CatalogFacts},
+    windmill_common::{datatable_roles_oss::datatable_roles_unavailable, error::Result},
+};
+
+/// Checked first by every ACL route, before anything is read or connected to.
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+pub(crate) fn ensure_datatable_acl_available() -> Result<()> {
+    Err(datatable_roles_unavailable())
+}
+
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+pub(crate) fn plan_statements(
+    _target: &AclTarget,
+    _change: &AclChange,
+    _dbname: &str,
+    _pg_role: &str,
+    _facts: &CatalogFacts,
+) -> Result<AclPlan> {
+    Err(datatable_roles_unavailable())
+}

backend/windmill-api-workspaces/src/datatable_migrations.rs

@@ -30,6 +30,7 @@ use windmill_api_auth::{require_super_admin, ApiAuthed};
 use windmill_api_jobs::run_wait_result_internal;
 use windmill_audit::audit_oss::audit_log;
 use windmill_audit::ActionKind;
+use windmill_common::datatable_roles::ADMIN_DATATABLE_ROLE;
 use windmill_common::db::UserDB;
 use windmill_common::error::{pg_error_message, Error, JsonResult, Result};
 use windmill_common::jobs::{JobPayload, RawCode};
@@ -38,7 +39,11 @@ use windmill_common::runnable_settings::{ConcurrencySettingsWithCustom, Debounci
 use windmill_common::scripts::ScriptLang;
 use windmill_common::users::username_to_permissioned_as;
 use windmill_common::worker::to_raw_value;
-use windmill_common::workspaces::get_datatable_resource_from_db_unchecked;
+use windmill_common::worker::SqlAnnotations;
+use windmill_common::workspaces::{
+    ensure_can_use_datatable_role, ensure_datatable_admin_access,
+    get_datatable_resource_from_db_unchecked, resolve_governing_datatable, DatatableAccess,
+};
 use windmill_common::{PgDatabase, DB};
 use windmill_git_sync::{
     handle_deployment_metadata, handle_deployment_metadata_batch, DeployedObject,
@@ -86,6 +91,42 @@ pub(crate) fn routes() -> Router {
         )
 }
 
+/// Refuse a migration whose role this caller may not use, before a job is pushed or a version
+/// recorded.
+///
+/// A migration that declares `-- role <name>` runs as that role, so the caller has to be one of its
+/// tenants. One that declares none runs as `admin` and reaches every object in the database
+/// whatever the roles grant, so it is for the admins of the workspace that governs the data table
+/// — a fork can run a migration under a role it holds, never a migration under `admin`.
+///
+/// The executor re-checks the role when it resolves the connection, so this is not the boundary. It
+/// is what makes the refusal legible: which migration, and which role.
+async fn ensure_migration_role_allowed(
+    db: &DB,
+    w_id: &str,
+    datatable_name: &str,
+    authed: &ApiAuthed,
+    sql: &str,
+    timestamp: i64,
+    name: &str,
+) -> Result<()> {
+    let context = format!("Migration {timestamp} ({name})");
+    let access = DatatableAccess::Authed(authed.to_authed_ref());
+    match SqlAnnotations::datatable_role(sql)? {
+        Some(role) => {
+            ensure_can_use_datatable_role(db, w_id, datatable_name, Some(&role), &access, &context)
+                .await
+        }
+        None => ensure_datatable_admin_access(db, w_id, datatable_name, &access)
+            .await
+            .map_err(|e| {
+                Error::NotAuthorized(format!(
+                    "{context} declares no role, so it would run as admin. {e}"
+                ))
+            }),
+    }
+}
+
 #[derive(Serialize)]
 struct AppliedMigration {
     version: i64,
@@ -128,7 +169,18 @@ async fn datatable_database_arg(
     .await?
     .ok_or_else(|| Error::internal_err(format!("datatable {datatable_name} not found")))?;
 
-    Ok(to_raw_value(&format!("datatable://{datatable_name}")))
+    // `?role=admin` rather than a bare reference, so a migration that declares no `-- role` runs
+    // as the connection that owns the schema instead of falling through to the data table's
+    // default role — which is what `ensure_migration_role_allowed` gated it as, and which is the
+    // only role a DDL statement can be expected to succeed under. A migration that does declare a
+    // role overrides this: the annotation wins over the reference.
+    //
+    // A legacy name containing `?` cannot be migrated through this reference: the appended query
+    // makes it neither an exact name nor a parseable one. Accepted on purpose, since such names can
+    // no longer be created and none are expected to carry migrations.
+    Ok(to_raw_value(&format!(
+        "datatable://{datatable_name}?role={ADMIN_DATATABLE_ROLE}"
+    )))
 }
 
 /// Run a migration's SQL as a normal Windmill `postgresql` job, permissioned as
@@ -384,6 +436,11 @@ async fn run_datatable_migrations(
     Path((w_id, datatable_name)): Path<(String, String)>,
     Query(query): Query<RunDatatableMigrationsQuery>,
 ) -> JsonResult<RunDatatableMigrationsResult> {
+    // Before the admin connection is opened at all: the bookkeeping below is created and read
+    // through it, so a caller no role covers must be refused here rather than after the fact.
+    crate::datatable_permissions::ensure_reaches_datatable(&db, &w_id, &datatable_name, &authed)
+        .await?;
+
     audit_log(
         &db,
         &authed,
@@ -440,6 +497,16 @@ async fn run_datatable_migrations(
         if applied_versions.contains(&m.timestamp) {
             continue;
         }
+        ensure_migration_role_allowed(
+            &db,
+            &w_id,
+            &datatable_name,
+            &authed,
+            &m.code_up,
+            m.timestamp,
+            &m.name,
+        )
+        .await?;
         run_datatable_migration_job(&db, &user_db, &authed, &w_id, &database_arg, &m.code_up)
             .await
             .map_err(|e| {
@@ -506,6 +573,11 @@ async fn rollback_datatable_migrations(
     Path((w_id, datatable_name)): Path<(String, String)>,
     Query(query): Query<RollbackDatatableMigrationsQuery>,
 ) -> JsonResult<RollbackDatatableMigrationsResult> {
+    // Before the admin connection is opened at all: the bookkeeping below is created and read
+    // through it, so a caller no role covers must be refused here rather than after the fact.
+    crate::datatable_permissions::ensure_reaches_datatable(&db, &w_id, &datatable_name, &authed)
+        .await?;
+
     audit_log(
         &db,
         &authed,
@@ -588,6 +660,17 @@ async fn rollback_datatable_migrations(
         ))
     })?;
 
+    ensure_migration_role_allowed(
+        &db,
+        &w_id,
+        &datatable_name,
+        &authed,
+        &code_down,
+        version,
+        &definition.name,
+    )
+    .await?;
+
     let database_arg = datatable_database_arg(&db, &w_id, &datatable_name).await?;
     run_datatable_migration_job(&db, &user_db, &authed, &w_id, &database_arg, &code_down)
         .await
@@ -748,10 +831,15 @@ async fn read_applied_datatable_versions(
 
 /// List a data table's migrations annotated with whether each has been applied.
 async fn datatable_migrations_status(
-    _authed: ApiAuthed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path((w_id, datatable_name)): Path<(String, String)>,
 ) -> JsonResult<DatatableMigrationsStatusResult> {
+    // Reads `_wm_migrations` through the data table's admin connection, so it answers to the same
+    // question as running one: may you reach this data table at all.
+    crate::datatable_permissions::ensure_reaches_datatable(&db, &w_id, &datatable_name, &authed)
+        .await?;
+
     let enabled = datatable_migrations_enabled(&db, &w_id, &datatable_name).await?;
     if !enabled {
         return Ok(Json(DatatableMigrationsStatusResult {
@@ -1431,6 +1519,15 @@ async fn generate_initial_datatable_migration(
     Extension(db): Extension<DB>,
     Path((w_id, datatable_name)): Path<(String, String)>,
 ) -> JsonResult<DatatableMigration> {
+    // Returns a `pg_dump` of the whole schema and writes into the data table's own bookkeeping, so
+    // it answers to the workspace that governs it rather than to whoever is asking.
+    ensure_datatable_admin_access(
+        &db,
+        &w_id,
+        &datatable_name,
+        &DatatableAccess::Authed(authed.to_authed_ref()),
+    )
+    .await?;
     validate_datatable_path_segment(&datatable_name)?;
     ensure_datatable_migrations_enabled(&db, &w_id, &datatable_name).await?;
 
@@ -1601,9 +1698,21 @@ pub(crate) struct DatatableRename {
     pub(crate) to: String,
 }
 
-async fn resolve_datatable_pg(db: &DB, w_id: &str, datatable: &str) -> Result<PgDatabase> {
+/// The database whose `_wm_migrations` a rename or delete of `datatable` in `w_id` should touch —
+/// `None` when that is somebody else's.
+///
+/// A fork's entry points at the workspace that governs the data table, so renaming or removing it
+/// changes what the fork calls the data table and nothing more. Following the pointer here would
+/// let a fork admin relabel or wipe the *governing* workspace's migration bookkeeping through
+/// their own settings form, and the parent would then re-run every migration from zero.
+async fn resolve_datatable_pg(db: &DB, w_id: &str, datatable: &str) -> Result<Option<PgDatabase>> {
+    let governing = resolve_governing_datatable(db, w_id, datatable).await?;
+    if governing.workspace_id != w_id {
+        return Ok(None);
+    }
     let db_resource = get_datatable_resource_from_db_unchecked(db, w_id, datatable).await?;
     serde_json::from_value(db_resource)
+        .map(Some)
         .map_err(|e| Error::internal_err(format!("Failed to parse database credentials: {}", e)))
 }
 
@@ -1621,7 +1730,9 @@ fn ignore_missing_wm_migrations(e: tokio_postgres::Error) -> Result<()> {
 
 /// Drop a data table's rows from its own database's `_wm_migrations`.
 async fn remote_forget_datatable_migrations(db: &DB, w_id: &str, datatable: &str) -> Result<()> {
-    let pg_db = resolve_datatable_pg(db, w_id, datatable).await?;
+    let Some(pg_db) = resolve_datatable_pg(db, w_id, datatable).await? else {
+        return Ok(());
+    };
     let (client, connection) = pg_db.connect(Some(db)).await?;
     tokio::spawn(async move {
         let _ = connection.await;
@@ -1646,7 +1757,9 @@ async fn remote_rename_datatable_migrations(
     from: &str,
     to: &str,
 ) -> Result<()> {
-    let pg_db = resolve_datatable_pg(db, w_id, resolve_by).await?;
+    let Some(pg_db) = resolve_datatable_pg(db, w_id, resolve_by).await? else {
+        return Ok(());
+    };
     let (client, connection) = pg_db.connect(Some(db)).await?;
     tokio::spawn(async move {
         let _ = connection.await;

backend/windmill-api-workspaces/src/datatable_permissions.rs

@@ -0,0 +1,79 @@
+/*
+ * Author: Ruben Fiszel
+ * Copyright: Windmill Labs, Inc 2022
+ * This file and its contents are licensed under the AGPLv3 License.
+ * Please see the included NOTICE for copyright information and
+ * LICENSE-AGPL for a copy of the license.
+ */
+
+//! Who may connect to a data table as which role.
+//!
+//! The decision lives on the data table entry of the workspace that governs it, which is not
+//! necessarily the workspace asking: a fork's entry points at its parent's, and everything here
+//! resolves through that pointer first. Nothing in this module runs SQL against the data table —
+//! a save is tenant lists and a default, and the Postgres roles themselves are the instance
+//! catalog's business.
+
+use axum::{routing::get, Router};
+
+use windmill_api_auth::ApiAuthed;
+use windmill_common::error::Result;
+use windmill_common::workspaces::GoverningDatatable;
+use windmill_common::DB;
+
+use crate::datatable_permissions_oss as roles;
+
+pub(crate) fn routes() -> Router {
+    Router::new()
+        .route(
+            "/datatable_permissions/{datatable_name}",
+            get(roles::get_datatable_permissions).post(roles::set_datatable_permissions),
+        )
+        .route(
+            "/datatable_usable_roles/{datatable_name}",
+            get(roles::list_usable_datatable_roles),
+        )
+}
+
+/// Administering a data table — its permissions, its migrations that declare no role, its exports
+/// — is for the admins of the workspace that governs it. A fork can use the data table; it never
+/// administers it.
+// The gate for whatever administers a data table under roles, which the routes of this module alone
+// do not always reach.
+#[allow(dead_code)]
+pub(crate) async fn ensure_governs_datatable(
+    db: &DB,
+    authed: &ApiAuthed,
+    w_id: &str,
+    governing: &GoverningDatatable,
+) -> Result<()> {
+    roles::ensure_governs_datatable(db, authed, w_id, governing).await
+}
+
+/// Refuse a caller that no tenant of this data table covers.
+///
+/// The bookkeeping endpoints below open the data table's `admin` connection to read or create
+/// `_wm_migrations` before they know which migration will run — so without this, someone covered
+/// by no role at all can still force admin-backed reads and writes on a database they may not
+/// touch. It asks only "may you reach this data table as anything"; which role a given migration
+/// runs as is still decided per migration, and by the executor after that.
+pub(crate) async fn ensure_reaches_datatable(
+    db: &DB,
+    w_id: &str,
+    datatable_name: &str,
+    authed: &ApiAuthed,
+) -> Result<()> {
+    roles::ensure_reaches_datatable(db, w_id, datatable_name, authed).await
+}
+
+/// [`ensure_reaches_datatable`] against an entry already resolved, for a caller that goes on to
+/// connect from that same entry.
+pub(crate) async fn ensure_reaches_governing_datatable(
+    db: &DB,
+    w_id: &str,
+    datatable_name: &str,
+    governing: &GoverningDatatable,
+    authed: &ApiAuthed,
+) -> Result<()> {
+    roles::ensure_reaches_governing_datatable(db, w_id, datatable_name, governing, authed).await
+}

backend/windmill-api-workspaces/src/datatable_permissions_oss.rs

@@ -0,0 +1,112 @@
+/*
+ * Author: Ruben Fiszel
+ * Copyright: Windmill Labs, Inc 2022
+ * This file and its contents are licensed under the AGPLv3 License.
+ * Please see the included NOTICE for copyright information and
+ * LICENSE-AGPL for a copy of the license.
+ */
+
+//! Where the data table permissions endpoints and their gates come from: the enterprise
+//! implementation, or a refusal. Roles are an Enterprise Edition feature; see
+//! `windmill_common::datatable_roles_oss`.
+
+#[cfg(all(feature = "private", feature = "enterprise"))]
+pub(crate) use crate::datatable_permissions_ee::{
+    ensure_governs_datatable, ensure_reaches_datatable, ensure_reaches_governing_datatable,
+    get_datatable_permissions, list_usable_datatable_roles, set_datatable_permissions,
+    usable_datatable_roles,
+};
+
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+pub(crate) use ce::*;
+
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+mod ce {
+    use windmill_api_auth::ApiAuthed;
+    use windmill_common::{
+        datatable_roles_oss::datatable_roles_unavailable as unavailable,
+        error::Result,
+        workspaces::{resolve_governing_datatable, GoverningDatatable},
+        DB,
+    };
+
+    /// Nobody administers a data table's roles without them.
+    #[allow(dead_code)]
+    pub(crate) async fn ensure_governs_datatable(
+        _db: &DB,
+        _authed: &ApiAuthed,
+        _w_id: &str,
+        _governing: &GoverningDatatable,
+    ) -> Result<()> {
+        Err(unavailable())
+    }
+
+    /// A data table not under roles is reached as it was before roles existed. One under roles is
+    /// refused: no role of it can be connected as.
+    pub(crate) async fn ensure_reaches_datatable(
+        db: &DB,
+        w_id: &str,
+        datatable_name: &str,
+        _authed: &ApiAuthed,
+    ) -> Result<()> {
+        let governing = resolve_governing_datatable(db, w_id, datatable_name).await?;
+        if governing.datatable.permissions.is_none() {
+            Ok(())
+        } else {
+            Err(unavailable())
+        }
+    }
+
+    pub(crate) async fn ensure_reaches_governing_datatable(
+        _db: &DB,
+        _w_id: &str,
+        _datatable_name: &str,
+        governing: &GoverningDatatable,
+        _authed: &ApiAuthed,
+    ) -> Result<()> {
+        if governing.datatable.permissions.is_none() {
+            Ok(())
+        } else {
+            Err(unavailable())
+        }
+    }
+
+    // The routes stay registered so the API has one shape; each answers after authentication,
+    // before anything is read.
+
+    pub(crate) async fn get_datatable_permissions(_authed: ApiAuthed) -> Result<String> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn set_datatable_permissions(_authed: ApiAuthed) -> Result<String> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn list_usable_datatable_roles(_authed: ApiAuthed) -> Result<String> {
+        Err(unavailable())
+    }
+
+    pub(crate) struct UsableDatatableRoles {
+        pub(crate) permissioned: bool,
+        pub(crate) roles: Vec<String>,
+        pub(crate) default_role: String,
+    }
+
+    /// A data table not under roles is used as `admin`, as before roles existed. One under roles
+    /// is refused: no role of it can be connected as.
+    pub(crate) async fn usable_datatable_roles(
+        _db: &DB,
+        _authed: &ApiAuthed,
+        _w_id: &str,
+        governing: &GoverningDatatable,
+    ) -> Result<UsableDatatableRoles> {
+        if governing.datatable.permissions.is_some() {
+            return Err(unavailable());
+        }
+        Ok(UsableDatatableRoles {
+            permissioned: false,
+            roles: vec![],
+            default_role: windmill_common::datatable_roles::ADMIN_DATATABLE_ROLE.to_string(),
+        })
+    }
+}

backend/windmill-api-workspaces/src/lib.rs

@@ -1,7 +1,11 @@
 #[cfg(feature = "parquet")]
 pub mod ai_session_backups;
 pub mod data_metrics;
+pub mod datatable_acl;
+pub mod datatable_acl_oss;
 pub mod datatable_migrations;
+pub mod datatable_permissions;
+pub mod datatable_permissions_oss;
 pub mod deployment_requests;
 pub mod workspaces;
 pub mod workspaces_extra;
@@ -9,3 +13,9 @@ pub mod workspaces_oss;
 
 #[cfg(feature = "private")]
 pub mod workspaces_ee;
+
+#[cfg(all(feature = "private", feature = "enterprise"))]
+pub mod datatable_acl_ee;
+
+#[cfg(all(feature = "private", feature = "enterprise"))]
+pub mod datatable_permissions_ee;

backend/windmill-api-workspaces/src/workspaces_extra.rs

@@ -492,6 +492,30 @@ pub(crate) async fn change_workspace_id(
     .fetch_all(&mut *tx)
     .await?;
 
+    // A fork's data table entry names the workspace that governs it by id, so the rename has to
+    // follow there too — anywhere, not just in the reparented children: a detached workspace can
+    // point at this one without being its fork. Left behind, the pointer resolves to the archived
+    // shell and every job through it stops.
+    info!("Re-pointing data table references to the new workspace id");
+    sqlx::query!(
+        r#"UPDATE workspace_settings ws
+           SET datatable = (
+               SELECT jsonb_set(ws.datatable, '{datatables}', jsonb_object_agg(
+                   dt.key,
+                   CASE WHEN dt.value->'reference'->>'workspace_id' = $2
+                       THEN jsonb_set(dt.value, '{reference,workspace_id}', to_jsonb($1::text))
+                       ELSE dt.value END
+               ))
+               FROM jsonb_each(ws.datatable->'datatables') dt
+           )
+           WHERE jsonb_typeof(ws.datatable->'datatables') = 'object'
+             AND ws.datatable::text LIKE '%"reference"%'"#,
+        &rw.new_id,
+        &old_id,
+    )
+    .execute(&mut *tx)
+    .await?;
+
     info!("Updating workspace_protection_rule table");
     sqlx::query!(
         "UPDATE workspace_protection_rule SET workspace_id = $1 WHERE workspace_id = $2",
@@ -971,6 +995,22 @@ pub(crate) async fn delete_workspace(
     // but the destructive cleanup itself runs only after the commit below: a delete that
     // fails mid-way must never leave a live workspace with its fork data destroyed and no
     // registry row to retry from. Read-only: nothing is dropped here.
+    // Read before the delete: another workspace's data table entry can point at one of this
+    // workspace's, and deleting the workspace it names leaves that pointer resolving to nothing.
+    // Nothing sweeps them — turning them back into copies would hand each fork the database
+    // outright — so the deleter is told which data tables they just stranded.
+    let stranded_pointers = sqlx::query!(
+        r#"SELECT ws.workspace_id AS "workspace_id!", dt.key AS "datatable!"
+           FROM workspace_settings ws
+           CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt
+           WHERE dt.value->'reference'->>'workspace_id' = $1
+           ORDER BY ws.workspace_id, dt.key"#,
+        &w_id,
+    )
+    .fetch_all(&db)
+    .await
+    .unwrap_or_default();
+
     let fork_ducklake_cleanups = prepare_fork_ducklake_cleanups(&db, &w_id, None)
         .await
         .unwrap_or_else(|e| {
@@ -1289,7 +1329,23 @@ pub(crate) async fn delete_workspace(
         tracing::warn!("failed to broadcast fork lineage change: {e:#}");
     }
 
-    Ok(format!("Deleted workspace {}", &w_id))
+    if stranded_pointers.is_empty() {
+        Ok(format!("Deleted workspace {}", &w_id))
+    } else {
+        let stranded = stranded_pointers
+            .iter()
+            .map(|r| format!("{}/{}", r.workspace_id, r.datatable))
+            .collect::<Vec<_>>()
+            .join(", ");
+        Ok(format!(
+            concat!(
+                "Deleted workspace {}. These data tables were governed by it and no longer ",
+                "resolve: {}. Their databases still exist; a superadmin can point them at ",
+                "another workspace's data table."
+            ),
+            &w_id, stranded
+        ))
+    }
 }
 
 #[derive(Deserialize)]
@@ -1343,15 +1399,20 @@ pub async fn drop_forked_datatable_databases(
     let mut errors: Vec<String> = Vec::new();
 
     for dt_name in &req.datatable_names {
-        let dt = match datatables.get(dt_name) {
-            Some(dt) if dt.forked_from.is_some() => dt,
+        // Only a clone is droppable, and a clone is terminal by construction: a kept data table is
+        // a pointer at the parent's database, which this fork does not own.
+        let database = match datatables.get(dt_name) {
+            Some(dt) if dt.forked_from.is_some() => match dt.database.as_ref() {
+                Some(database) => database,
+                None => continue,
+            },
             _ => continue,
         };
 
-        if dt.database.resource_type
+        if database.resource_type
             == windmill_common::workspaces::DataTableCatalogResourceType::Instance
         {
-            let db_to_drop = &dt.database.resource_path;
+            let db_to_drop = &database.resource_path;
             if !db_to_drop.starts_with("wm_fork_") {
                 errors.push(format!(
                     "Refusing to drop instance database '{}' for datatable://{}:  name does not start with 'wm_fork_'",

backend/windmill-api/openapi.yaml

@@ -1572,6 +1572,43 @@ paths:
               schema:
                 type: object
 
+  /settings/external_instance_pg/status:
+    get:
+      summary: Returns whether the external instance cluster is configured and how its last setup went
+      operationId: getExternalInstancePgStatus
+      tags:
+        - setting
+      responses:
+        "200":
+          description: external instance cluster status
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ExternalInstancePgStatus"
+
+  /settings/external_instance_pg/setup:
+    post:
+      summary: Sets up the external instance cluster with its saved admin login, optionally rotating the passwords Windmill manages on it (enterprise edition only)
+      operationId: setupExternalInstancePg
+      tags:
+        - setting
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                rotate_passwords:
+                  type: boolean
+      responses:
+        "200":
+          description: the setup report, also stored as the last setup
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ExternalInstancePgSetupReport"
+
   /settings/list_custom_instance_pg_databases:
     post:
       summary: Returns the set-up statuses of custom instance pg databases
@@ -1588,6 +1625,89 @@ paths:
                 additionalProperties:
                   $ref: "#/components/schemas/CustomInstanceDb"
 
+  /settings/datatable_roles:
+    get:
+      summary: list the instance's data table roles
+      operationId: listInstanceDatatableRoles
+      tags:
+        - setting
+      responses:
+        "200":
+          description: the instance role catalog
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: "#/components/schemas/InstanceDatatableRole"
+    post:
+      summary: create a data table role on the instance's Postgres cluster
+      operationId: createInstanceDatatableRole
+      tags:
+        - setting
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              required: [name]
+              properties:
+                name:
+                  type: string
+      responses:
+        "200":
+          description: the created role
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/InstanceDatatableRole"
+
+  /settings/datatable_roles/{id}:
+    post:
+      summary: rename a data table role or turn its login on and off
+      operationId: updateInstanceDatatableRole
+      tags:
+        - setting
+      parameters:
+        - in: path
+          name: id
+          required: true
+          schema:
+            type: string
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                name:
+                  type: string
+                enabled:
+                  type: boolean
+      responses:
+        "200":
+          description: the updated role
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/InstanceDatatableRole"
+    delete:
+      summary: drop a data table role from the cluster and from every workspace that named it
+      operationId: deleteInstanceDatatableRole
+      tags:
+        - setting
+      parameters:
+        - in: path
+          name: id
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          description: deleted
+
   /settings/setup_custom_instance_pg_database/{name}:
     post:
       summary: Runs CREATE DATABASE on the Windmill Postgres and grants access to the custom_instance_user
@@ -5141,7 +5261,7 @@ paths:
                 type: array
                 items:
                   type: object
-                  required: [name, resource_type, resource_path]
+                  required: [name, resource_type, resource_path, permissioned]
                   properties:
                     name:
                       type: string
@@ -5150,6 +5270,188 @@ paths:
                       enum: [postgres, instance]
                     resource_path:
                       type: string
+                    governing_workspace_id:
+                      type: string
+                    permissioned:
+                      type: boolean
+
+  /w/{workspace}/workspaces/datatable_permissions/{datatable_name}:
+    get:
+      summary: get who may connect to a data table as which role
+      operationId: getDatatablePermissions
+      tags:
+        - workspace
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - name: datatable_name
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          description: the data table's roles and their tenants
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/DatatablePermissions"
+    post:
+      summary: set who may connect to a data table as which role
+      operationId: setDatatablePermissions
+      tags:
+        - workspace
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - name: datatable_name
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              required: [permissioned]
+              properties:
+                permissioned:
+                  type: boolean
+                default_role:
+                  type: string
+                roles:
+                  type: array
+                  items:
+                    $ref: "#/components/schemas/DatatableRoleTenants"
+      responses:
+        "200":
+          description: status
+          content:
+            text/plain:
+              schema:
+                type: string
+
+  /w/{workspace}/workspaces/datatable_acl/{datatable_name}:
+    get:
+      summary: read the owner and grants of an instance data table's database, schema or table
+      operationId: getDatatableAcl
+      tags:
+        - workspace
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - name: datatable_name
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: kind
+          in: query
+          required: true
+          schema:
+            type: string
+            enum: [database, schema, table]
+        - name: schema
+          in: query
+          required: false
+          schema:
+            type: string
+        - name: table
+          in: query
+          required: false
+          schema:
+            type: string
+      responses:
+        "200":
+          description: owner and grants
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/DatatableAclInfo"
+
+  /w/{workspace}/workspaces/datatable_acl/{datatable_name}/plan:
+    post:
+      summary: preview the SQL an ownership or grant change would run (data table administrators only)
+      operationId: planDatatableAcl
+      tags:
+        - workspace
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - name: datatable_name
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/AclChangeRequest"
+      responses:
+        "200":
+          description: statements that would run, in a single transaction
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AclPlan"
+
+  /w/{workspace}/workspaces/datatable_acl/{datatable_name}/apply:
+    post:
+      summary: run an ownership or grant change exactly as planned (data table administrators only)
+      operationId: applyDatatableAcl
+      tags:
+        - workspace
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - name: datatable_name
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/AclChangeRequest"
+      responses:
+        "200":
+          description: change applied
+          content:
+            text/plain:
+              schema:
+                type: string
+
+  /w/{workspace}/workspaces/datatable_usable_roles/{datatable_name}:
+    get:
+      summary: list the data table roles the caller may connect as
+      operationId: listUsableDatatableRoles
+      tags:
+        - workspace
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - name: datatable_name
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          description: usable roles
+          content:
+            application/json:
+              schema:
+                type: object
+                required: [permissioned, roles, default_role]
+                properties:
+                  permissioned:
+                    type: boolean
+                  roles:
+                    type: array
+                    items:
+                      type: string
+                  default_role:
+                    type: string
 
   /w/{workspace}/workspaces/list_datatable_schemas:
     get:
@@ -5223,6 +5525,21 @@ paths:
         - workspace
       parameters:
         - $ref: "#/components/parameters/WorkspaceId"
+        - name: datatable_name
+          in: query
+          description: list only this data table; each listed data table opens a connection to its database
+          schema:
+            type: string
+        - name: role_for
+          in: query
+          description: the data table `role` applies to; every other one is listed as its default role
+          schema:
+            type: string
+        - name: role
+          in: query
+          description: the role to list `role_for` as; refused, in that entry's `error`, if the caller may not use it
+          schema:
+            type: string
       responses:
         "200":
           description: table metadata of all datatables
@@ -5256,6 +5573,11 @@ paths:
           required: true
           schema:
             type: string
+        - name: role
+          in: query
+          description: the data table role to read the table as; defaults to the data table's default role
+          schema:
+            type: string
       responses:
         "200":
           description: schema of one datatable table
@@ -5331,7 +5653,22 @@ paths:
           description: status
           content:
             application/json:
-              schema: {}
+              schema:
+                type: object
+                properties:
+                  stranded_references:
+                    description: >-
+                      Data tables in other workspaces that were governed by one this save deleted
+                      and no longer resolve.
+                    type: array
+                    items:
+                      type: object
+                      required: [workspace_id, datatable]
+                      properties:
+                        workspace_id:
+                          type: string
+                        datatable:
+                          type: string
 
   /w/{workspace}/workspaces/run_datatable_migrations/{datatable_name}:
     post:
@@ -33321,6 +33658,353 @@ components:
         - ducklake
         - datatable
 
+    ExternalInstancePgSetupStep:
+      type: object
+      required: [name, status, message]
+      properties:
+        name:
+          type: string
+        status:
+          type: string
+          enum: [ok, warning, error]
+        message:
+          type: string
+
+    ExternalInstancePgSetupReport:
+      type: object
+      required: [success, finished_at, steps]
+      properties:
+        success:
+          type: boolean
+          description: no step failed; warnings leave it true
+        finished_at:
+          type: string
+          format: date-time
+        steps:
+          type: array
+          items:
+            $ref: "#/components/schemas/ExternalInstancePgSetupStep"
+
+    ExternalInstancePgStatus:
+      type: object
+      required: [configured, database_count]
+      properties:
+        configured:
+          type: boolean
+        database_count:
+          type: integer
+        last_setup:
+          $ref: "#/components/schemas/ExternalInstancePgSetupReport"
+
+    InstanceDatatableRole:
+      type: object
+      required: [id, name, enabled]
+      properties:
+        id:
+          type: string
+        name:
+          type: string
+        enabled:
+          type: boolean
+
+    DatatableRoleTenants:
+      type: object
+      required: [id, tenants]
+      properties:
+        id:
+          type: string
+        name:
+          type: string
+        tenants:
+          type: array
+          items:
+            type: string
+
+    DatatablePermissions:
+      type: object
+      required: [supported, permissioned, default_role, roles, editable, available_roles]
+      properties:
+        supported:
+          type: boolean
+          description: >-
+            Whether this data table can be put under roles at all. Only one backed by the
+            instance database can: a role is a login on that cluster.
+        permissioned:
+          type: boolean
+        default_role:
+          type: string
+        roles:
+          type: array
+          items:
+            $ref: "#/components/schemas/DatatableRoleTenants"
+        governing_workspace_id:
+          type: string
+        editable:
+          type: boolean
+        available_roles:
+          type: array
+          items:
+            $ref: "#/components/schemas/InstanceDatatableRole"
+        ungoverned_reachers:
+          type: array
+          items:
+            type: object
+            required: [workspace_id, datatable]
+            properties:
+              workspace_id:
+                type: string
+              datatable:
+                type: string
+
+    AclTarget:
+      description: what access is read or changed on
+      oneOf:
+        - $ref: "#/components/schemas/AclTargetDatabase"
+        - $ref: "#/components/schemas/AclTargetSchema"
+        - $ref: "#/components/schemas/AclTargetTable"
+      discriminator:
+        propertyName: kind
+        mapping:
+          database: "#/components/schemas/AclTargetDatabase"
+          schema: "#/components/schemas/AclTargetSchema"
+          table: "#/components/schemas/AclTargetTable"
+
+    AclTargetDatabase:
+      type: object
+      required: [kind]
+      properties:
+        kind:
+          type: string
+          enum: [database]
+
+    AclTargetSchema:
+      type: object
+      required: [kind, schema]
+      properties:
+        kind:
+          type: string
+          enum: [schema]
+        schema:
+          type: string
+
+    AclTargetTable:
+      type: object
+      required: [kind, schema, table]
+      properties:
+        kind:
+          type: string
+          enum: [table]
+        schema:
+          type: string
+        table:
+          type: string
+
+    AclChange:
+      description: one change to plan or apply
+      oneOf:
+        - $ref: "#/components/schemas/AclChangeSetOwner"
+        - $ref: "#/components/schemas/AclChangeGrant"
+        - $ref: "#/components/schemas/AclChangeRevoke"
+      discriminator:
+        propertyName: type
+        mapping:
+          set_owner: "#/components/schemas/AclChangeSetOwner"
+          grant: "#/components/schemas/AclChangeGrant"
+          revoke: "#/components/schemas/AclChangeRevoke"
+
+    AclChangeSetOwner:
+      type: object
+      description: >-
+        hands the target to role — for a schema, with everything already in it but an extension's
+        members, which stay with the extension
+      required: [type, role]
+      properties:
+        type:
+          type: string
+          enum: [set_owner]
+        role:
+          type: string
+          description: a data table role of the instance, or admin
+
+    AclChangeGrant:
+      type: object
+      required: [type, role, privileges, scope]
+      properties:
+        type:
+          type: string
+          enum: [grant]
+        role:
+          type: string
+          description: a data table role of the instance, or admin
+        privileges:
+          type: array
+          items:
+            type: string
+        scope:
+          $ref: "#/components/schemas/AclGrantScope"
+
+    AclChangeRevoke:
+      type: object
+      required: [type, role, privileges, scope]
+      properties:
+        type:
+          type: string
+          enum: [revoke]
+        role:
+          type: string
+          description: a data table role of the instance, other than admin
+        privileges:
+          type: array
+          items:
+            type: string
+        scope:
+          $ref: "#/components/schemas/AclGrantScope"
+        objects:
+          type: array
+          description: >-
+            objects inside the target the revoke covers, empty for the target itself. Only with the
+            target scope; a revoke on all objects of a kind is refused, since it cannot say which
+            grants it takes back.
+          items:
+            $ref: "#/components/schemas/AclObject"
+
+    AclGrantScope:
+      type: string
+      enum:
+        [
+          target,
+          all_tables,
+          all_sequences,
+          all_functions,
+          future_tables,
+          future_sequences,
+          future_functions,
+        ]
+
+    AclChangeRequest:
+      type: object
+      required: [target, change]
+      properties:
+        target:
+          $ref: "#/components/schemas/AclTarget"
+        change:
+          $ref: "#/components/schemas/AclChange"
+        statements:
+          type: array
+          description: >-
+            The statements the plan showed. Required to apply, which plans again and refuses if
+            the result differs.
+          items:
+            type: string
+
+    AclPlan:
+      type: object
+      required: [statements, warnings]
+      properties:
+        statements:
+          type: array
+          items:
+            type: string
+        warnings:
+          type: array
+          items:
+            type: string
+
+    AclObject:
+      type: object
+      required: [name, kind]
+      properties:
+        name:
+          type: string
+        kind:
+          type: string
+          description: >-
+            TABLE, SEQUENCE, FUNCTION, PROCEDURE or TYPE — what the object is. A revoke turns it
+            into the keyword it takes, ROUTINE for both routine kinds; a type's grants are read
+            only.
+        args:
+          type: string
+          description: identity arguments of a routine, which is what tells two of the same name apart
+
+    AclGrant:
+      type: object
+      required: [grantee, privileges, sources]
+      properties:
+        grantee:
+          type: string
+        privileges:
+          type: array
+          items:
+            type: string
+        object:
+          $ref: "#/components/schemas/AclObject"
+        future:
+          type: string
+          description: >-
+            set for a default privilege, naming the kind of object it covers (TABLES, SEQUENCES,
+            FUNCTIONS, TYPES, or SCHEMAS). On a schema, the defaults set in that schema; on the
+            database, the ones set database-wide, which apply in every schema and which no schema's
+            own defaults take back.
+        sources:
+          type: array
+          description: >-
+            the roles the grant comes from, each once — who granted it, or for a default privilege
+            the role whose future objects it covers. A revoke of some of the grant's privileges
+            takes them back from every source that gave them.
+          items:
+            $ref: "#/components/schemas/AclSource"
+
+    AclSource:
+      type: object
+      required: [role, privileges, reachable]
+      properties:
+        role:
+          type: string
+        privileges:
+          type: array
+          description: >-
+            what role gave of the grant's privileges. A revoke is held back only by a source out of
+            reach that gave some of what it takes back.
+          items:
+            type: string
+        reachable:
+          type: boolean
+          description: >-
+            whether the data table's connection can take back what role gave. On an object that is
+            the owner, when the connection acts for the owner, and otherwise the connection itself;
+            for a default privilege, a creating role the connection acts for. What a source out of
+            reach gave is not revocable from here; privileges only other sources gave still are.
+
+    DatatableAclInfo:
+      type: object
+      required: [owner, roles, editable, supports_maintain, dbname, grants, children]
+      properties:
+        owner:
+          type: string
+        roles:
+          type: array
+          description: the roles a change may name; empty unless the caller may change anything
+          items:
+            type: string
+        editable:
+          type: boolean
+          description: whether the caller may plan and apply changes
+        supports_maintain:
+          type: boolean
+          description: whether the server is Postgres 17+, which added the MAINTAIN table privilege
+        dbname:
+          type: string
+          description: the database the target lives in
+        grants:
+          type: array
+          items:
+            $ref: "#/components/schemas/AclGrant"
+        children:
+          type: array
+          description: a database's schemas, or a schema's tables
+          items:
+            type: string
+
     CustomInstanceDb:
       type: object
       required:
@@ -35369,9 +36053,11 @@ components:
           type: object
           additionalProperties:
             type: object
-            required: [database]
             properties:
               database:
+                description: >-
+                  Set on an entry that owns its database. Absent on a fork's entry, which points at
+                  another workspace's data table instead.
                 type: object
                 properties:
                   resource_type:
@@ -35383,6 +36069,17 @@ components:
                     type: string
                 required:
                   - resource_type
+              reference:
+                description: >-
+                  The workspace and data table that govern this one. Server-owned: written by fork
+                  creation, and carried across a settings save whatever the request says.
+                type: object
+                required: [workspace_id, datatable]
+                properties:
+                  workspace_id:
+                    type: string
+                  datatable:
+                    type: string
               migrations_enabled:
                 type: boolean
                 description: Whether the SQL migrations feature is opted in for this data table
@@ -35451,7 +36148,17 @@ components:
 
     DataTableTables:
       type: object
-      required: [datatable_name, schemas]
+      required:
+        [
+          datatable_name,
+          schemas,
+          instance,
+          permissioned,
+          usable_roles,
+          default_role,
+          can_create_schema,
+          creatable_schemas,
+        ]
       properties:
         datatable_name:
           type: string
@@ -35464,6 +36171,26 @@ components:
               type: string
         error:
           type: string
+        instance:
+          type: boolean
+          description: on the instance database, the only kind that can be under roles or have its access edited
+        permissioned:
+          type: boolean
+        usable_roles:
+          type: array
+          description: the roles the caller may connect as, by name; empty when not under roles
+          items:
+            type: string
+        default_role:
+          type: string
+        can_create_schema:
+          type: boolean
+          description: whether the role the listing connected as may create schemas
+        creatable_schemas:
+          type: array
+          description: the schemas the role the listing connected as may create in
+          items:
+            type: string
 
     DataTableTableSchema:
       type: object

backend/windmill-api/src/capture.rs

@@ -567,6 +567,9 @@ async fn set_config(
     };
 
     let mut tx = user_db.begin(&authed).await?;
+    if matches!(nc.trigger_kind, TriggerKind::Postgres) {
+        windmill_common::datatable_roles::lock_datatable_streams(&mut *tx, false).await?;
+    }
 
     sqlx::query!(
         r#"
@@ -614,6 +617,9 @@ async fn ping_config(
     )>,
 ) -> Result<()> {
     let mut tx = user_db.begin(&authed).await?;
+    if matches!(trigger_kind, TriggerKind::Postgres) {
+        windmill_common::datatable_roles::lock_datatable_streams(&mut *tx, false).await?;
+    }
 
     sqlx::query!(
         r#"

backend/windmill-api/src/jobs.rs

@@ -8343,8 +8343,9 @@ pub async fn run_wait_result_flow_by_version(
 /// job lives, in particular DuckDB, which runs in-process in the worker.
 ///
 /// What it does permit is any statement against the workspace's data tables, writes and DDL
-/// included: the helper's body is an unrestricted SQL template and data tables carry no
-/// per-user ACL. Narrowing that is a separate decision from this exemption.
+/// included: the helper's body is an unrestricted SQL template. What that reaches is the
+/// operator's own data table role — the preview job is permissioned as them, so the executor
+/// resolves it under their tenancy like any other job.
 ///
 /// The database argument is only half the target: the executor honors a `-- database`
 /// directive in the SQL over it, and `-- s3` redirects the result set, so both are refused.
@@ -8576,7 +8577,7 @@ async fn run_inline_preview_script(
 #[cfg(not(feature = "run_inline"))]
 async fn run_inline_preview_script() -> error::Result<Response> {
     Err(error::Error::InternalErr(
-        "inline preview requires the worker feature".to_string(),
+        "inline preview requires the run_inline feature on the worker".to_string(),
     ))
 }
 
@@ -8700,7 +8701,12 @@ fn register_potential_assets_on_inline_execution(
             .as_ref()
             .and_then(|args| args.get("database"))
             .map(|v| v.get().trim_matches('"'))
-            .and_then(|dt| dt.strip_prefix("datatable://"));
+            .and_then(|dt| dt.strip_prefix("datatable://"))
+            // `?role=` picks the connection, not the data table. Anything else after a `?` may be
+            // part of a name stored before names were restricted, so it stays.
+            .map(|dt| {
+                windmill_common::workspaces::parse_datatable_ref(dt).map_or(dt, |(name, _)| name)
+            });
         if let Some(datatable) = datatable {
             let re = regex::Regex::new(r#"SET search_path TO "([^"]+)";"#).unwrap();
             let (schema, content) = if let Some(captures) = re.captures(&preview.content) {
@@ -9553,6 +9559,7 @@ async fn run_preview_flow_job(
             &flow_path,
             &run_query,
             user_message.as_ref(),
+            uuid,
         )
         .await?;
     }

backend/windmill-api/src/users.rs

@@ -352,6 +352,17 @@ async fn update_username_in_workpsace<'c>(
     new_username: &str,
     w_id: &str,
 ) -> error::Result<()> {
+    // ---- data table tenants ----
+    // Tenants name the user, so the rename has to follow here too; a list left naming the old
+    // username silently drops the access instead of moving it.
+    windmill_common::workspaces::rename_datatable_tenant_in_workspace(
+        tx,
+        w_id,
+        &format!("u/{old_username}"),
+        &format!("u/{new_username}"),
+    )
+    .await?;
+
     // ---- instance and workspace users ----
     sqlx::query!(
         "UPDATE usr SET username = $1 WHERE email = $2",

backend/windmill-api/src/workspaces_export.rs

@@ -1639,7 +1639,7 @@ pub(crate) async fn tarball_workspace(
                 mute_critical_alerts: row.mute_critical_alerts,
                 color: row.color.clone(),
                 operator_settings: row.operator_settings.clone(),
-                datatable: row.datatable.clone(),
+                datatable: windmill_common::workspaces::strip_datatable_permissions(row.datatable.clone()),
                 slack_team_id: row.slack_team_id.clone(),
                 slack_name: row.slack_name.clone(),
                 slack_command_script: row.slack_command_script.clone(),
@@ -1703,7 +1703,7 @@ pub(crate) async fn tarball_workspace(
                 mute_critical_alerts: row.mute_critical_alerts,
                 color: row.color,
                 operator_settings: row.operator_settings,
-                datatable: row.datatable,
+                datatable: windmill_common::workspaces::strip_datatable_permissions(row.datatable),
                 slack_team_id: row.slack_team_id,
                 slack_name: row.slack_name,
                 slack_command_script: row.slack_command_script,

backend/windmill-common/Cargo.toml

@@ -75,6 +75,7 @@ bitflags.workspace = true
 once_cell.workspace = true
 phf.workspace = true
 tokio-postgres.workspace = true
+postgres-protocol.workspace = true
 postgres-native-tls.workspace = true
 native-tls.workspace = true
 

backend/windmill-common/src/datatable_roles.rs

@@ -0,0 +1,359 @@
+/*
+ * Author: Ruben Fiszel
+ * Copyright: Windmill Labs, Inc 2022
+ * This file and its contents are licensed under the AGPLv3 License.
+ * Please see the included NOTICE for copyright information and
+ * LICENSE-AGPL for a copy of the license.
+ */
+
+//! The instance's data table role catalog.
+//!
+//! A data table role is a real Postgres login role on the Windmill cluster, named exactly as the
+//! user named it, shared by every instance database. Windmill decides who may ask for a role (the
+//! per-data-table tenant lists in [`crate::workspaces`]); Postgres decides what the role may then
+//! touch. The catalog here is only the first half's vocabulary plus the cluster provisioning.
+//!
+//! Entries are keyed by a generated id so a rename moves nothing else: tenants name the id.
+
+use std::collections::BTreeMap;
+
+use crate::{
+    error::{Error, Result},
+    DB,
+};
+
+/// The connection every data table resolved to before roles existed (`custom_instance_user`). It
+/// owns every pre-existing object, so it is a reserved name rather than a catalog entry: never
+/// created, renamed or dropped.
+pub const ADMIN_DATATABLE_ROLE: &str = "admin";
+
+/// The login the admin connection uses, and the role every created role is granted to — that
+/// membership is what later lets it `ALTER ... OWNER TO` a role and drop it.
+pub const CUSTOM_INSTANCE_USER: &str = "custom_instance_user";
+
+/// One catalog entry, as stored in `datatable_role`. The password is per role and instance-wide;
+/// it belongs to the instance, not to any workspace's settings.
+/// No `Serialize`/`Deserialize`: the catalog is rows now, and a derived `Serialize` would emit
+/// `pwd` — the same way out for a credential that the hand-written `Debug` below closes on the log
+/// side.
+#[derive(Clone)]
+pub struct InstanceDatatableRole {
+    /// The Postgres role name, verbatim.
+    pub name: String,
+    pub enabled: bool,
+    /// Absent only for a role whose provisioning did not finish; resolving as it then errors
+    /// rather than falling back to admin.
+    ///
+    /// A plain string rather than a `StringOrSecretRef` like the instance user's password: that
+    /// one is a secret ref because an operator supplies it and may want it to come from their own
+    /// backend, while this one is minted here and never entered by anyone, so there is nothing for
+    /// a ref to point at. Encrypting generated secrets at rest is a separate change that would
+    /// take the replication password with it.
+    pub pwd: Option<String>,
+}
+
+/// Hand-written so `{:?}` on a catalog cannot put a live Postgres password in a log line or an
+/// audit record. Everything else about the entry is safe to print.
+impl std::fmt::Debug for InstanceDatatableRole {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("InstanceDatatableRole")
+            .field("name", &self.name)
+            .field("enabled", &self.enabled)
+            .field("pwd", &self.pwd.as_ref().map(|_| "<redacted>"))
+            .finish()
+    }
+}
+
+pub type DatatableRoleCatalog = BTreeMap<String, InstanceDatatableRole>;
+
+/// Names Postgres or Windmill already owns. `admin` is excluded because it never reaches the
+/// cluster as a role name at all — it resolves to `custom_instance_user`.
+fn is_reserved_role_name(name: &str) -> bool {
+    let lower = name.to_ascii_lowercase();
+    lower == ADMIN_DATATABLE_ROLE
+        || lower == "postgres"
+        || lower == "public"
+        || lower.starts_with("pg_")
+        || lower.starts_with("windmill_")
+        || lower.starts_with("custom_instance_")
+}
+
+/// The charset is what makes every downstream interpolation safe: the name reaches Postgres as a
+/// quoted identifier, a `-- role <name>` annotation, and a `?role=` query parameter.
+pub fn validate_role_name(name: &str) -> Result<()> {
+    if name.is_empty() || name.len() > 63 {
+        return Err(Error::BadRequest(format!(
+            "Invalid data table role name '{name}': it must be between 1 and 63 characters"
+        )));
+    }
+    if !name
+        .chars()
+        .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
+    {
+        return Err(Error::BadRequest(format!(
+            "Invalid data table role name '{name}': only letters, digits, '_' and '-' are allowed"
+        )));
+    }
+    if is_reserved_role_name(name) {
+        return Err(Error::BadRequest(format!(
+            "'{name}' is reserved and cannot be used as a data table role name"
+        )));
+    }
+    Ok(())
+}
+
+/// A double-quoted Postgres identifier. Doubling `"` is Postgres's own escaping inside one, so this
+/// quotes any name — schema, table or role. Role names are validated as well
+/// ([`validate_role_name`]) because they also travel unquoted, in `-- role <name>` and `?role=`.
+pub fn quote_ident(name: &str) -> String {
+    format!("\"{}\"", name.replace('"', "\"\""))
+}
+
+/// Serialize the mutations that are not already serialized by the row itself.
+///
+/// A create is an insert and a delete is a delete, which Postgres orders for us — the unique index
+/// on `name` is what makes two concurrent creates of the same name one winner and one error. What
+/// still needs it is the window between the cluster DDL and the row: `CREATE ROLE` is not visible
+/// to another transaction's `pg_roles` check until commit, so without this two creates of the same
+/// name both pass their existence check and one fails on the index having already made the login.
+/// Held for the transaction, so the DDL has to run on that same transaction to be covered.
+pub async fn lock_role_catalog(tx: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> Result<()> {
+    sqlx::query!("SELECT pg_advisory_xact_lock(hashtext('datatable_role_catalog'))")
+        .execute(&mut **tx)
+        .await?;
+    Ok(())
+}
+
+/// A replication stream reads every row whatever a data table's roles grant. Turning roles on looks
+/// for streams holding this exclusive; whatever can start a Postgres trigger or capture streaming
+/// holds it shared on the transaction that commits it. So either the look sees the stream, or the
+/// stream's listener connects after roles are committed and refuses. Held for the transaction.
+pub async fn lock_datatable_streams(conn: &mut sqlx::PgConnection, exclusive: bool) -> Result<()> {
+    let lock = if exclusive {
+        "pg_advisory_xact_lock"
+    } else {
+        "pg_advisory_xact_lock_shared"
+    };
+    sqlx::query(&format!("SELECT {lock}(hashtext('datatable_streams'))"))
+        .execute(conn)
+        .await?;
+    Ok(())
+}
+
+/// Whether an instance database is reached only through entries under roles is decided by two
+/// writes that lock different workspaces' settings rows: turning roles on for one entry, and a
+/// settings save pointing an entry without roles at the database. Each holds this for every
+/// database it decides on, so neither reads past the other's uncommitted write. Held for the
+/// transaction; the names are locked in sorted order so two holders cannot deadlock.
+pub async fn lock_instance_databases_governance<'a>(
+    conn: &mut sqlx::PgConnection,
+    dbnames: impl IntoIterator<Item = &'a str>,
+) -> Result<()> {
+    let dbnames: std::collections::BTreeSet<&str> = dbnames.into_iter().collect();
+    for dbname in dbnames {
+        sqlx::query("SELECT pg_advisory_xact_lock(hashtext('datatable_instance_database:' || $1))")
+            .bind(dbname)
+            .execute(&mut *conn)
+            .await?;
+    }
+    Ok(())
+}
+
+/// Disclosure: returns every role's stored Postgres password in plaintext. Any server path that
+/// has to resolve or name a role may call it — including handlers open to a workspace member, who
+/// need the names — but callers MUST NOT let `pwd` reach a response, a log line, an audit record
+/// or an export. Nothing about who may call it: the credential is the whole risk, and `Debug` is
+/// hand-written to redact it for the same reason.
+pub async fn read_role_catalog(db: &DB) -> Result<DatatableRoleCatalog> {
+    crate::datatable_roles_oss::read_role_catalog(db).await
+}
+
+/// As [`read_role_catalog`], reading inside the caller's transaction so the value is the one
+/// [`lock_role_catalog`] is protecting. Same disclosure contract.
+pub async fn read_role_catalog_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+) -> Result<DatatableRoleCatalog> {
+    crate::datatable_roles_oss::read_role_catalog_tx(tx).await
+}
+
+/// Record a role, in the caller's transaction so it commits with the `CREATE ROLE` it describes.
+///
+/// Authorization: writes a generated Postgres credential. Callers MUST restrict this to superadmin
+/// paths and MUST hold [`lock_role_catalog`] on `tx`.
+pub async fn insert_role_catalog_entry(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    id: &str,
+    role: &InstanceDatatableRole,
+) -> Result<()> {
+    crate::datatable_roles_oss::insert_role_catalog_entry(tx, id, role).await
+}
+
+/// Update a role's recorded name, login flag and password. Same contract as
+/// [`insert_role_catalog_entry`].
+pub async fn update_role_catalog_entry(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    id: &str,
+    role: &InstanceDatatableRole,
+) -> Result<()> {
+    crate::datatable_roles_oss::update_role_catalog_entry(tx, id, role).await
+}
+
+/// Forget a role. Same contract as [`insert_role_catalog_entry`]; run it in the transaction that
+/// drops the cluster login, so the two cannot disagree.
+pub async fn delete_role_catalog_entry(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    id: &str,
+) -> Result<()> {
+    crate::datatable_roles_oss::delete_role_catalog_entry(tx, id).await
+}
+
+/// Resolve the role a caller named to its catalog id. A disabled role is an error rather than a
+/// silent fallback: the caller asked for something the instance deliberately turned off.
+pub fn role_id_by_name<'a>(catalog: &'a DatatableRoleCatalog, name: &str) -> Result<&'a str> {
+    let entry = catalog
+        .iter()
+        .find(|(_, role)| role.name == name)
+        .ok_or_else(|| {
+            Error::NotFound(format!(
+                "'{name}' is not a data table role of this instance. Defined roles: {}.",
+                catalog
+                    .values()
+                    .map(|r| r.name.as_str())
+                    .collect::<Vec<_>>()
+                    .join(", ")
+            ))
+        })?;
+    if !entry.1.enabled {
+        return Err(Error::BadRequest(format!(
+            "Data table role '{name}' is disabled on this instance"
+        )));
+    }
+    Ok(entry.0.as_str())
+}
+
+/// Every instance database the registry knows about. Role provisioning has to reach all of them:
+/// a role that cannot `CONNECT` to a database is refused by Postgres before any grant matters.
+pub async fn registered_instance_databases(db: &DB) -> Result<Vec<String>> {
+    crate::datatable_roles_oss::registered_instance_databases(db).await
+}
+
+/// `CONNECT` on `dbname` for every enabled role, and none for `PUBLIC`. Run at role creation, at
+/// database creation, and lazily whenever an instance data table is administered, so a database
+/// provisioned before a role existed is repaired rather than left silently unreachable.
+///
+/// Authorization: rewrites a database's ACL with the server's own credentials and checks nothing.
+/// Callers MUST have authorized administration of `dbname` — superadmin, or an admin of the
+/// workspace governing a data table on it.
+pub async fn converge_connect_grants(db: &DB, dbname: &str) -> Result<()> {
+    crate::datatable_roles_oss::converge_connect_grants(db, dbname).await
+}
+
+/// As [`converge_connect_grants`], with a catalog the caller already read. Same contract.
+pub async fn converge_connect_grants_with(
+    db: &DB,
+    dbname: &str,
+    catalog: &DatatableRoleCatalog,
+) -> Result<()> {
+    crate::datatable_roles_oss::converge_connect_grants_with(db, dbname, catalog).await
+}
+
+/// `CREATE ROLE <name> LOGIN PASSWORD ...; GRANT <name> TO custom_instance_user`, and `CONNECT` on
+/// every registered database. No privileges beyond that — an admin grants them through SQL or the
+/// ACL editor.
+///
+/// Authorization: creates a cluster-wide Postgres login. Callers MUST restrict this to superadmin
+/// paths, and MUST hold [`lock_role_catalog`] on the same transaction.
+pub async fn create_instance_role(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    name: &str,
+    password: &str,
+) -> Result<()> {
+    crate::datatable_roles_oss::create_instance_role(tx, name, password).await
+}
+
+/// Authorization: alters a cluster-wide Postgres login. Callers MUST restrict this to superadmin
+/// paths, and MUST hold [`lock_role_catalog`] on the same transaction.
+pub async fn set_instance_role_login(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    name: &str,
+    enabled: bool,
+) -> Result<()> {
+    crate::datatable_roles_oss::set_instance_role_login(tx, name, enabled).await
+}
+
+/// A rename discards an md5-hashed password, so the caller has to hand over a fresh one.
+///
+/// Authorization: renames a cluster-wide Postgres login. Callers MUST restrict this to superadmin
+/// paths, and MUST hold [`lock_role_catalog`] on the same transaction.
+pub async fn rename_instance_role(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    from: &str,
+    to: &str,
+    password: &str,
+) -> Result<()> {
+    crate::datatable_roles_oss::rename_instance_role(tx, from, to, password).await
+}
+
+/// A role owning anything in any database blocks its own `DROP ROLE`, and both its objects and the
+/// privileges granted to it are only visible from inside each database — hence the pass over the
+/// registry. An unreachable database aborts the whole delete: dropping the role while one database
+/// still holds objects owned by it leaves those objects owned by a numeric OID nobody can name.
+///
+/// Each pass runs as the instance's own Postgres user rather than `custom_instance_user`, which
+/// owns the databases and can therefore revoke a grant whoever made it. `custom_instance_user`
+/// could only undo what it granted itself, so a privilege planted by an operator in psql — the
+/// ordinary way privileges reach a role — would survive and block the drop.
+///
+/// Authorization: drops a cluster-wide Postgres login and reassigns everything it owns. Callers
+/// MUST restrict this to superadmin paths, and MUST hold [`lock_role_catalog`] on `tx`.
+///
+/// The per-database passes open their own connections and cannot join `tx`; the lock is what keeps
+/// a concurrent mutation out while they run. Only the final `DROP ROLE` is on `tx`, so it commits
+/// or rolls back with the catalog write that forgets the role. Those passes commit as they go, so
+/// callers MUST have disabled the role in an earlier committed transaction: a failure part-way
+/// then leaves a disabled role to retry, not an enabled one already stripped in some databases.
+pub async fn drop_instance_role(
+    db: &DB,
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    name: &str,
+) -> Result<()> {
+    crate::datatable_roles_oss::drop_instance_role(db, tx, name).await
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn role_names_are_validated() {
+        assert!(validate_role_name("analytics").is_ok());
+        assert!(validate_role_name("read-only_2").is_ok());
+        assert!(validate_role_name("").is_err());
+        assert!(validate_role_name(&"a".repeat(64)).is_err());
+        assert!(validate_role_name("has space").is_err());
+        assert!(validate_role_name("quote\"injection").is_err());
+        // Reserved, case-insensitively.
+        assert!(validate_role_name("admin").is_err());
+        assert!(validate_role_name("Postgres").is_err());
+        assert!(validate_role_name("pg_read_all_data").is_err());
+        assert!(validate_role_name("windmill_user").is_err());
+        assert!(validate_role_name("custom_instance_user").is_err());
+    }
+
+    #[test]
+    fn a_disabled_role_is_an_error_not_a_fallback() {
+        let mut catalog = DatatableRoleCatalog::new();
+        catalog.insert(
+            "id1".to_string(),
+            InstanceDatatableRole {
+                name: "analytics".to_string(),
+                enabled: false,
+                pwd: Some("x".to_string()),
+            },
+        );
+        assert!(role_id_by_name(&catalog, "analytics").is_err());
+        assert!(role_id_by_name(&catalog, "nope").is_err());
+        catalog.get_mut("id1").unwrap().enabled = true;
+        assert_eq!(role_id_by_name(&catalog, "analytics").unwrap(), "id1");
+    }
+}

backend/windmill-common/src/datatable_roles_oss.rs

@@ -0,0 +1,209 @@
+/*
+ * Author: Ruben Fiszel
+ * Copyright: Windmill Labs, Inc 2022
+ * This file and its contents are licensed under the AGPLv3 License.
+ * Please see the included NOTICE for copyright information and
+ * LICENSE-AGPL for a copy of the license.
+ */
+
+//! Where data table roles come from: the enterprise implementation, or a refusal.
+//!
+//! Roles are an Enterprise Edition feature. An edition without them creates, grants and connects
+//! as none, and a data table saved under roles — by an enterprise build, before a downgrade — is
+//! refused rather than resolved as `admin`. A data table not under roles, asked for no role,
+//! resolves as it always has. `private` alone is not that edition: community builds carry it.
+
+use crate::error::Error;
+
+/// What every roles path answers without the Enterprise Edition. The frontend matches this exact
+/// sentence (`datatableUsableRoles.ts`) to read the refusal as "not under roles": reword both.
+pub fn datatable_roles_unavailable() -> Error {
+    Error::BadRequest("Data table roles are a Windmill Enterprise Edition feature".to_string())
+}
+
+#[cfg(all(feature = "private", feature = "enterprise"))]
+pub(crate) use crate::datatable_roles_ee::{
+    can_use_datatable_role, can_use_datatable_role_in_governing_workspace, converge_connect_grants,
+    converge_connect_grants_with, create_instance_role, delete_role_catalog_entry,
+    drop_instance_role, ensure_can_use_datatable_role, ensure_datatable_admin_access,
+    ensure_instance_db_grant_options_unchecked, forget_datatable_role_everywhere,
+    insert_role_catalog_entry, read_role_catalog, read_role_catalog_tx,
+    registered_instance_databases, rename_instance_role, resolve_datatable_role_connection,
+    set_instance_role_login, update_role_catalog_entry,
+};
+
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+pub(crate) use ce::*;
+
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+mod ce {
+    use super::datatable_roles_unavailable as unavailable;
+    use crate::{
+        datatable_roles::{DatatableRoleCatalog, InstanceDatatableRole},
+        db::AuthedRef,
+        error::Result,
+        workspaces::{
+            resolve_governing_datatable, DataTableRoleTenants, DatatableAccess, GoverningDatatable,
+        },
+        DB,
+    };
+
+    type Tx<'a> = sqlx::Transaction<'a, sqlx::Postgres>;
+
+    pub(crate) async fn read_role_catalog(_db: &DB) -> Result<DatatableRoleCatalog> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn read_role_catalog_tx(_tx: &mut Tx<'_>) -> Result<DatatableRoleCatalog> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn insert_role_catalog_entry(
+        _tx: &mut Tx<'_>,
+        _id: &str,
+        _role: &InstanceDatatableRole,
+    ) -> Result<()> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn update_role_catalog_entry(
+        _tx: &mut Tx<'_>,
+        _id: &str,
+        _role: &InstanceDatatableRole,
+    ) -> Result<()> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn delete_role_catalog_entry(_tx: &mut Tx<'_>, _id: &str) -> Result<()> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn registered_instance_databases(_db: &DB) -> Result<Vec<String>> {
+        Err(unavailable())
+    }
+
+    /// Nothing to converge: with no roles to admit, an instance database keeps the `CONNECT`
+    /// grants it was created with, `PUBLIC`'s included, as it did before roles existed.
+    pub(crate) async fn converge_connect_grants(_db: &DB, _dbname: &str) -> Result<()> {
+        Ok(())
+    }
+
+    /// As [`converge_connect_grants`].
+    pub(crate) async fn converge_connect_grants_with(
+        _db: &DB,
+        _dbname: &str,
+        _catalog: &DatatableRoleCatalog,
+    ) -> Result<()> {
+        Ok(())
+    }
+
+    pub(crate) async fn create_instance_role(
+        _tx: &mut Tx<'_>,
+        _name: &str,
+        _password: &str,
+    ) -> Result<()> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn set_instance_role_login(
+        _tx: &mut Tx<'_>,
+        _name: &str,
+        _enabled: bool,
+    ) -> Result<()> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn rename_instance_role(
+        _tx: &mut Tx<'_>,
+        _from: &str,
+        _to: &str,
+        _password: &str,
+    ) -> Result<()> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn drop_instance_role(_db: &DB, _tx: &mut Tx<'_>, _name: &str) -> Result<()> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn ensure_instance_db_grant_options_unchecked(
+        _db: &DB,
+        _dbname: &str,
+    ) -> Result<()> {
+        Err(unavailable())
+    }
+
+    /// No tenant list covers anyone: there is no role to connect as.
+    pub(crate) fn can_use_datatable_role(
+        _tenants: &DataTableRoleTenants,
+        _authed: &AuthedRef<'_>,
+    ) -> bool {
+        false
+    }
+
+    pub(crate) async fn can_use_datatable_role_in_governing_workspace(
+        _db: &DB,
+        _governing_w_id: &str,
+        _w_id: &str,
+        _tenants: &DataTableRoleTenants,
+        _access: &DatatableAccess<'_>,
+    ) -> Result<bool> {
+        Err(unavailable())
+    }
+
+    /// Reached only for a data table under roles or a caller naming a role: both are refused.
+    pub(crate) async fn resolve_datatable_role_connection(
+        _db: &DB,
+        _w_id: &str,
+        _name: &str,
+        _governing: &GoverningDatatable,
+        _db_resource: serde_json::Value,
+        _role: Option<&str>,
+        _access: DatatableAccess<'_>,
+    ) -> Result<serde_json::Value> {
+        Err(unavailable())
+    }
+
+    /// A data table not under roles, asked for no role or for `admin`, is not a role decision and
+    /// passes, as it did before roles existed. Anything else is refused.
+    pub(crate) async fn ensure_can_use_datatable_role(
+        db: &DB,
+        w_id: &str,
+        name: &str,
+        role: Option<&str>,
+        _access: &DatatableAccess<'_>,
+        _context: &str,
+    ) -> Result<()> {
+        let governing = resolve_governing_datatable(db, w_id, name).await?;
+        if governing.datatable.permissions.is_none()
+            && role.is_none_or(|r| r == crate::datatable_roles::ADMIN_DATATABLE_ROLE)
+        {
+            Ok(())
+        } else {
+            Err(unavailable())
+        }
+    }
+
+    /// A data table not under roles is the `admin` connection for anyone who reaches it, as before
+    /// roles existed. One under roles is refused.
+    pub(crate) async fn ensure_datatable_admin_access(
+        db: &DB,
+        w_id: &str,
+        name: &str,
+        _access: &DatatableAccess<'_>,
+    ) -> Result<()> {
+        let governing = resolve_governing_datatable(db, w_id, name).await?;
+        if governing.datatable.permissions.is_none() {
+            Ok(())
+        } else {
+            Err(unavailable())
+        }
+    }
+
+    pub(crate) async fn forget_datatable_role_everywhere(
+        _tx: &mut Tx<'_>,
+        _role_id: &str,
+    ) -> Result<()> {
+        Err(unavailable())
+    }
+}

backend/windmill-common/src/external_instance_pg.rs

@@ -0,0 +1,178 @@
+/*
+ * Author: Ruben Fiszel
+ * Copyright: Windmill Labs, Inc 2022
+ * This file and its contents are licensed under the AGPLv3 License.
+ * Please see the included NOTICE for copyright information and
+ * LICENSE-AGPL for a copy of the license.
+ */
+
+//! The external Postgres cluster behind `external_instance` data tables and Ducklake catalogs.
+//!
+//! Windmill administers that cluster itself, logged in as the user in
+//! [`EXTERNAL_INSTANCE_PG_SETTING`]. It creates `custom_instance_user` and
+//! `custom_instance_replication_user` there, with passwords it generates and keeps in
+//! [`EXTERNAL_INSTANCE_PG_STATE_SETTING`]. They share their names with the roles on Windmill's own
+//! cluster, but they are different roles with different passwords.
+//!
+//! The cluster may hold data Windmill did not create. Two Windmill instances sharing one is not
+//! supported: each would keep resetting the passwords the other depends on.
+
+use std::collections::BTreeMap;
+
+use serde::{Deserialize, Serialize};
+
+use crate::{
+    error::{Error, Result},
+    global_settings::{EXTERNAL_INSTANCE_PG_SETTING, EXTERNAL_INSTANCE_PG_STATE_SETTING},
+    instance_config::{CustomInstanceDb, ExternalInstancePg},
+    DB,
+};
+
+/// What Windmill keeps about the external cluster. Server-managed and hidden: never part of the
+/// instance config, never readable by an agent worker. No `Debug`: it carries live passwords.
+#[derive(Serialize, Deserialize, Clone, Default)]
+pub struct ExternalInstancePgState {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub user_pwd: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub replication_pwd: Option<String>,
+    /// The databases Windmill created on the cluster. It only ever drops one of these.
+    #[serde(default)]
+    pub databases: BTreeMap<String, CustomInstanceDb>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub last_setup: Option<ExternalInstancePgSetupReport>,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct ExternalInstancePgSetupReport {
+    /// No step failed. Warnings leave it true.
+    pub success: bool,
+    pub finished_at: chrono::DateTime<chrono::Utc>,
+    pub steps: Vec<ExternalInstancePgSetupStep>,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct ExternalInstancePgSetupStep {
+    pub name: String,
+    pub status: SetupStepStatus,
+    pub message: String,
+}
+
+#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq)]
+#[serde(rename_all = "lowercase")]
+pub enum SetupStepStatus {
+    Ok,
+    Warning,
+    Error,
+}
+
+/// The status the settings page shows without running anything.
+#[derive(Serialize, Debug)]
+pub struct ExternalInstancePgStatus {
+    pub configured: bool,
+    pub database_count: usize,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub last_setup: Option<ExternalInstancePgSetupReport>,
+}
+
+/// Authorization: returns the cluster's admin password and checks nothing. Callers MUST be
+/// superadmin or an internal server path.
+pub(crate) async fn read_external_instance_pg_config<'c>(
+    executor: impl sqlx::PgExecutor<'c>,
+) -> Result<Option<ExternalInstancePg>> {
+    let value = sqlx::query_scalar!(
+        "SELECT value FROM global_settings WHERE name = $1",
+        EXTERNAL_INSTANCE_PG_SETTING
+    )
+    .fetch_optional(executor)
+    .await?;
+    value
+        .map(|v| {
+            serde_json::from_value(v).map_err(|e| {
+                Error::internal_err(format!("reading {EXTERNAL_INSTANCE_PG_SETTING}: {e}"))
+            })
+        })
+        .transpose()
+}
+
+/// Authorization: returns the passwords Windmill generated on the cluster and checks nothing.
+/// Callers MUST be superadmin or an internal server path.
+pub(crate) async fn read_external_instance_pg_state<'c>(
+    executor: impl sqlx::PgExecutor<'c>,
+) -> Result<ExternalInstancePgState> {
+    let value = sqlx::query_scalar!(
+        "SELECT value FROM global_settings WHERE name = $1",
+        EXTERNAL_INSTANCE_PG_STATE_SETTING
+    )
+    .fetch_optional(executor)
+    .await?;
+    match value {
+        None => Ok(ExternalInstancePgState::default()),
+        Some(v) => serde_json::from_value(v).map_err(|e| {
+            Error::internal_err(format!("reading {EXTERNAL_INSTANCE_PG_STATE_SETTING}: {e}"))
+        }),
+    }
+}
+
+pub async fn external_instance_pg_status(db: &DB) -> Result<ExternalInstancePgStatus> {
+    let configured = read_external_instance_pg_config(db).await?.is_some();
+    let state = read_external_instance_pg_state(db).await?;
+    Ok(ExternalInstancePgStatus {
+        configured,
+        database_count: state.databases.len(),
+        last_setup: state.last_setup,
+    })
+}
+
+/// Refuse to unset the cluster while Windmill still has databases on it: every data table and
+/// Ducklake catalog there would stop resolving. Allowed on every edition, so a downgraded
+/// instance can still clear a setting it no longer uses.
+pub async fn ensure_external_instance_pg_removable(db: &DB) -> Result<()> {
+    let state = read_external_instance_pg_state(db).await?;
+    if state.databases.is_empty() {
+        return Ok(());
+    }
+    let names = state
+        .databases
+        .keys()
+        .cloned()
+        .collect::<Vec<_>>()
+        .join(", ");
+    Err(Error::BadRequest(format!(
+        "The external instance cluster still holds databases Windmill created ({names}). Drop \
+         them before removing {EXTERNAL_INSTANCE_PG_SETTING}."
+    )))
+}
+
+/// Check a write to [`EXTERNAL_INSTANCE_PG_SETTING`] before it happens: `None`, null or an empty
+/// string unsets it. Every writer of global settings calls this, the per-key and bulk endpoints
+/// as well as the declarative sync.
+pub async fn check_external_instance_pg_write(
+    db: &DB,
+    value: Option<&serde_json::Value>,
+) -> Result<()> {
+    match value {
+        None | Some(serde_json::Value::Null) => ensure_external_instance_pg_removable(db).await,
+        Some(serde_json::Value::String(s)) if s.trim().is_empty() => {
+            ensure_external_instance_pg_removable(db).await
+        }
+        Some(value) => {
+            crate::external_instance_pg_oss::validate_external_instance_pg_setting(value)
+        }
+    }
+}
+
+/// Converge the external cluster on the configured login: check what it can do, create or update
+/// Windmill's two roles with the stored passwords, and report anything that would get in the way.
+/// With `rotate_passwords`, generate new passwords first. Safe to run again; running it again is
+/// how a failed rotation is repaired.
+///
+/// Authorization: administers the external cluster with its admin credentials and checks nothing.
+/// Callers MUST be superadmin.
+pub async fn setup_external_instance_pg_unchecked(
+    db: &DB,
+    rotate_passwords: bool,
+) -> Result<ExternalInstancePgSetupReport> {
+    crate::external_instance_pg_oss::setup_external_instance_pg_unchecked(db, rotate_passwords)
+        .await
+}

backend/windmill-common/src/external_instance_pg_oss.rs

@@ -0,0 +1,43 @@
+/*
+ * Author: Ruben Fiszel
+ * Copyright: Windmill Labs, Inc 2022
+ * This file and its contents are licensed under the AGPLv3 License.
+ * Please see the included NOTICE for copyright information and
+ * LICENSE-AGPL for a copy of the license.
+ */
+
+//! Where the external instance cluster comes from: the enterprise implementation, or a refusal.
+//! `private` alone is not that edition: community builds carry it.
+
+use crate::error::Error;
+
+pub fn external_instance_pg_unavailable() -> Error {
+    Error::BadRequest(
+        "External instance databases are a Windmill Enterprise Edition feature".to_string(),
+    )
+}
+
+#[cfg(all(feature = "private", feature = "enterprise"))]
+pub(crate) use crate::external_instance_pg_ee::{
+    setup_external_instance_pg_unchecked, validate_external_instance_pg_setting,
+};
+
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+pub(crate) use ce::*;
+
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+mod ce {
+    use super::external_instance_pg_unavailable as unavailable;
+    use crate::{error::Result, external_instance_pg::ExternalInstancePgSetupReport, DB};
+
+    pub(crate) fn validate_external_instance_pg_setting(_value: &serde_json::Value) -> Result<()> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn setup_external_instance_pg_unchecked(
+        _db: &DB,
+        _rotate_passwords: bool,
+    ) -> Result<ExternalInstancePgSetupReport> {
+        Err(unavailable())
+    }
+}

backend/windmill-common/src/flow_conversations.rs

@@ -36,30 +36,20 @@ pub async fn get_or_create_conversation_with_id(
     title: &str,
     conversation_id: Uuid,
 ) -> Result<FlowConversation> {
-    // Check if conversation already exists
-    let existing_conversation = sqlx::query_as!(
-        FlowConversation,
-        "SELECT id, workspace_id, flow_path, title, created_at, updated_at, created_by
-         FROM flow_conversation
-         WHERE id = $1 AND workspace_id = $2",
-        conversation_id,
-        w_id
-    )
-    .fetch_optional(&mut **tx)
-    .await?;
-
-    if let Some(existing) = existing_conversation {
+    if let Some(existing) = lock_conversation(tx, w_id, conversation_id).await? {
         return Ok(existing);
     }
 
     // Truncate title to 25 characters max
     let title = truncate_with_ellipsis(title, 25);
 
-    // Create new conversation with provided ID
-    let conversation = sqlx::query_as!(
+    // Every turn released by the same collector's commit finds no row: the first insert
+    // wins, the others wait on it, do nothing, and read the row it created.
+    let created = sqlx::query_as!(
         FlowConversation,
         "INSERT INTO flow_conversation (id, workspace_id, flow_path, created_by, title)
          VALUES ($1, $2, $3, $4, $5)
+         ON CONFLICT (id) DO NOTHING
          RETURNING id, workspace_id, flow_path, title, created_at, updated_at, created_by",
         conversation_id,
         w_id,
@@ -67,10 +57,41 @@ pub async fn get_or_create_conversation_with_id(
         username,
         title
     )
-    .fetch_one(&mut **tx)
+    .fetch_optional(&mut **tx)
     .await?;
+    if let Some(conversation) = created {
+        return Ok(conversation);
+    }
 
-    Ok(conversation)
+    lock_conversation(tx, w_id, conversation_id)
+        .await?
+        .ok_or_else(|| {
+            crate::error::Error::BadRequest(format!(
+                "conversation {conversation_id} belongs to another workspace"
+            ))
+        })
+}
+
+/// Locked, so a turn orders against retention collecting the conversation
+/// (windmill_common::jobs::delete_jobs): either the turn goes first and the collector then
+/// sees its message, or it waits and finds the row gone and creates it again. Unlocked, the
+/// message insert would wait on the parent row's lock instead and then fail its FK check.
+async fn lock_conversation(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    w_id: &str,
+    conversation_id: Uuid,
+) -> Result<Option<FlowConversation>> {
+    Ok(sqlx::query_as!(
+        FlowConversation,
+        "SELECT id, workspace_id, flow_path, title, created_at, updated_at, created_by
+         FROM flow_conversation
+         WHERE id = $1 AND workspace_id = $2
+         FOR UPDATE",
+        conversation_id,
+        w_id
+    )
+    .fetch_optional(&mut **tx)
+    .await?)
 }
 
 /// Add a message to a conversation using an existing transaction

backend/windmill-common/src/global_settings.rs

@@ -57,6 +57,8 @@ pub const SAML_METADATA_SETTING: &str = "saml_metadata";
 pub const SMTP_SETTING: &str = "smtp_settings";
 pub const TEAMS_SETTING: &str = "teams";
 pub const INDEXER_SETTING: &str = "indexer_settings";
+pub const EXTERNAL_INSTANCE_PG_SETTING: &str = "external_instance_pg";
+pub const EXTERNAL_INSTANCE_PG_STATE_SETTING: &str = "external_instance_pg_state";
 pub const TIMEOUT_WAIT_RESULT_SETTING: &str = "timeout_wait_result";
 
 pub const UNIQUE_ID_SETTING: &str = "uid";
@@ -355,6 +357,9 @@ pub const AGENT_WORKER_BLOCKED_SETTINGS: &[&str] = &[
     // resolve datatable connections through the dedicated datatable endpoints, never these.
     "custom_instance_pg_databases",
     "custom_instance_replication_pwd",
+    // The external cluster's admin login, and the passwords Windmill generated on it.
+    EXTERNAL_INSTANCE_PG_SETTING,
+    EXTERNAL_INSTANCE_PG_STATE_SETTING,
 ];
 
 /// Whether an agent worker may read the given global setting over HTTP.

backend/windmill-common/src/instance_config.rs

@@ -350,6 +350,8 @@ pub struct GlobalSettings {
     pub ducklake_settings: Option<DucklakeSettings>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub custom_instance_pg_databases: Option<CustomInstancePgDatabases>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub external_instance_pg: Option<ExternalInstancePg>,
 
     // Opaque settings (EE-private structs or no clear schema)
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -833,6 +835,36 @@ pub struct CustomInstanceDbLogs {
     pub replication_user_error: Option<String>,
 }
 
+// ---------------------------------------------------------------------------
+// External instance PG cluster
+// ---------------------------------------------------------------------------
+
+/// The external Postgres cluster Windmill manages for `external_instance` data tables and Ducklake
+/// catalogs. `user` logs in as the cluster's administrator: it needs `CREATEDB` and `CREATEROLE`.
+/// `dbname` is only where that login connects to run cluster-wide statements.
+///
+/// Every field defaults rather than being required: this deserializes as part of the whole
+/// instance config, and one malformed row must not make every other setting unreadable. The
+/// write path and every use reject an incomplete value instead.
+#[derive(Deserialize, Serialize, Clone, Debug, Default)]
+#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
+pub struct ExternalInstancePg {
+    #[serde(default)]
+    pub host: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub port: Option<u16>,
+    #[serde(default)]
+    pub user: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub password: Option<StringOrSecretRef>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub dbname: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub sslmode: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub root_certificate_pem: Option<String>,
+}
+
 // ---------------------------------------------------------------------------
 // Autoscaling (worker config)
 // ---------------------------------------------------------------------------
@@ -969,6 +1001,7 @@ pub const PROTECTED_SETTINGS: &[&str] = &[
     "ducklake_settings",
     "custom_instance_pg_databases",
     "custom_instance_replication_pwd",
+    "external_instance_pg_state",
     "uid",
     "rsa_keys",
     "jwt_secret",
@@ -994,6 +1027,8 @@ pub const HIDDEN_SETTINGS: &[&str] = &[
     // Server-only (written by setup/refresh via direct SQL), never operator-authored —
     // hidden so the config machinery can't read, rewrite, or drop it.
     "custom_instance_replication_pwd",
+    // Same for the passwords and database registry Windmill keeps for the external cluster.
+    "external_instance_pg_state",
 ];
 
 /// Top-level settings whose entire value is sensitive and must be fully redacted in logs.
@@ -1005,6 +1040,7 @@ const SENSITIVE_SETTINGS: &[&str] = &[
     "license_key",
     "ducklake_user_pg_pwd",
     "custom_instance_replication_pwd",
+    "external_instance_pg_state",
     "pip_index_url",
     "pip_extra_index_url",
     "npm_config_registry",
@@ -1030,6 +1066,7 @@ const NESTED_SENSITIVE_FIELDS: &[(&str, &[&str])] = &[
         &["secret_key", "serviceAccountKey"],
     ),
     ("custom_instance_pg_databases", &["user_pwd"]),
+    ("external_instance_pg", &["password"]),
 ];
 
 fn redact_json_value(value: &serde_json::Value) -> serde_json::Value {
@@ -1359,6 +1396,13 @@ pub async fn sync_global_settings_declarative(
         .map_err(|e| anyhow::anyhow!("{origins_key}: {e}"))?;
 
     let diff = diff_global_settings(current, desired, ApplyMode::Replace);
+    let external_pg_key = crate::global_settings::EXTERNAL_INSTANCE_PG_SETTING;
+    if diff.deletes.iter().any(|k| k == external_pg_key) {
+        crate::external_instance_pg::check_external_instance_pg_write(db, None).await?;
+    }
+    if let Some(value) = diff.upserts.get(external_pg_key) {
+        crate::external_instance_pg::check_external_instance_pg_write(db, Some(value)).await?;
+    }
     apply_settings_diff(db, &diff).await?;
 
     Ok(())
@@ -1491,6 +1535,10 @@ pub fn resolve_env_refs(settings: &mut GlobalSettings) -> Result<(), String> {
         resolve_env_option(&mut pg.user_pwd)?;
     }
 
+    if let Some(pg) = &mut settings.external_instance_pg {
+        resolve_env_option(&mut pg.password)?;
+    }
+
     Ok(())
 }
 
@@ -2460,39 +2508,33 @@ mod tests {
     }
 
     #[test]
-    fn custom_instance_replication_pwd_is_isolated_from_config() {
-        // The replication-role password is server-only: written by setup/refresh via direct
-        // SQL, never operator-authored. It must stay out of the declarative config surface
-        // (hidden on read) and be undeletable, so config sync can't read, rewrite, or drop it.
-        assert!(HIDDEN_SETTINGS.contains(&"custom_instance_replication_pwd"));
-        assert!(PROTECTED_SETTINGS.contains(&"custom_instance_replication_pwd"));
-        assert!(SENSITIVE_SETTINGS.contains(&"custom_instance_replication_pwd"));
+    fn server_generated_db_passwords_are_isolated_from_config() {
+        // These hold passwords the server generates: written by setup/refresh via direct SQL,
+        // never operator-authored. They must stay out of the declarative config surface
+        // (hidden on read) and be undeletable, so config sync can't read, rewrite, or drop them.
+        for key in [
+            "custom_instance_replication_pwd",
+            "external_instance_pg_state",
+        ] {
+            assert!(HIDDEN_SETTINGS.contains(&key), "{key}");
+            assert!(PROTECTED_SETTINGS.contains(&key), "{key}");
+            assert!(SENSITIVE_SETTINGS.contains(&key), "{key}");
 
-        // A stray desired value (e.g. flattened into `extra`) is ignored, not upserted.
-        let mut desired = BTreeMap::new();
-        desired.insert(
-            "custom_instance_replication_pwd".to_string(),
-            serde_json::json!("attacker-set"),
-        );
-        let diff = diff_global_settings(&BTreeMap::new(), &desired, ApplyMode::Merge);
-        assert!(
-            diff.upserts.is_empty(),
-            "hidden setting must not be upserted"
-        );
+            // A stray desired value (e.g. flattened into `extra`) is ignored, not upserted.
+            let mut desired = BTreeMap::new();
+            desired.insert(key.to_string(), serde_json::json!("attacker-set"));
+            let diff = diff_global_settings(&BTreeMap::new(), &desired, ApplyMode::Merge);
+            assert!(diff.upserts.is_empty(), "{key} must not be upserted");
 
-        // A current value is never deleted by a Replace that omits it.
-        let mut current = BTreeMap::new();
-        current.insert(
-            "custom_instance_replication_pwd".to_string(),
-            serde_json::json!("live"),
-        );
-        let diff = diff_global_settings(&current, &BTreeMap::new(), ApplyMode::Replace);
-        assert!(
-            !diff
-                .deletes
-                .contains(&"custom_instance_replication_pwd".to_string()),
-            "hidden setting must not be deleted"
-        );
+            // A current value is never deleted by a Replace that omits it.
+            let mut current = BTreeMap::new();
+            current.insert(key.to_string(), serde_json::json!("live"));
+            let diff = diff_global_settings(&current, &BTreeMap::new(), ApplyMode::Replace);
+            assert!(
+                !diff.deletes.contains(&key.to_string()),
+                "{key} must not be deleted"
+            );
+        }
     }
 
     #[test]

backend/windmill-common/src/jobs.rs

@@ -478,6 +478,12 @@ pub static WORKER_INTERNAL_SERVER_INLINE_UTILS: OnceCell<WorkerInternalServerInl
 /// set-based deletes below cost one scan per table per call instead. Because the cascade no
 /// longer fires, every code path that deletes from `v2_job` by id must go through this helper
 /// (or delete these tables itself) or it will leave orphan rows behind.
+/// **Transaction contract:** call this inside a transaction. The conversation cleanup below
+/// locks rows to serialise itself against a concurrent delete, and on an autocommit
+/// connection that lock is released at statement end, silently restoring the race.
+/// A conversation is collected only once every message row of it has gone with a job; a
+/// row written with no job id (an MCP tool call, persisted under no job of its own) keeps
+/// its conversation and the agent's memory for it alive for as long as it exists.
 pub async fn delete_jobs(conn: &mut sqlx::PgConnection, ids: &[uuid::Uuid]) -> error::Result<()> {
     sqlx::query!(
         "DELETE FROM dispatch_event WHERE producer_job_id = ANY($1)",
@@ -485,12 +491,55 @@ pub async fn delete_jobs(conn: &mut sqlx::PgConnection, ids: &[uuid::Uuid]) -> e
     )
     .execute(&mut *conn)
     .await?;
-    sqlx::query!(
-        "DELETE FROM flow_conversation_message WHERE job_id = ANY($1)",
+    let mut conversation_ids: Vec<uuid::Uuid> = sqlx::query_scalar!(
+        "DELETE FROM flow_conversation_message WHERE job_id = ANY($1) RETURNING conversation_id",
         ids
     )
-    .execute(&mut *conn)
+    .fetch_all(&mut *conn)
     .await?;
+    conversation_ids.sort_unstable();
+    conversation_ids.dedup();
+    if !conversation_ids.is_empty() {
+        // A conversation is a view over its messages: once the last one goes with its job,
+        // the row and the agent's memory for it are all that is left, and nothing else
+        // collects them — `ai_agent_memory` carries no job id for retention to match on.
+        // Two statements rather than one CTE: a data-modifying CTE reads the snapshot from
+        // before the delete above, so every conversation would still look non-empty.
+        // Two calls each deleting one of a conversation's last messages would each still see
+        // the other's row — uncommitted deletes are invisible across transactions — so
+        // neither would collect it and nothing would try again. Taking the conversation row
+        // first serialises them: the second reads the first's delete and finds it empty.
+        sqlx::query_scalar!(
+            "SELECT id FROM flow_conversation WHERE id = ANY($1) ORDER BY id FOR UPDATE",
+            &conversation_ids
+        )
+        .fetch_all(&mut *conn)
+        .await?;
+        // Memory first, since it reads the conversation row for its workspace.
+        sqlx::query!(
+            "DELETE FROM ai_agent_memory a
+               USING flow_conversation c
+              WHERE c.id = ANY($1)
+                AND a.conversation_id = c.id
+                AND a.workspace_id = c.workspace_id
+                AND NOT EXISTS (
+                    SELECT 1 FROM flow_conversation_message m WHERE m.conversation_id = c.id
+                )",
+            &conversation_ids
+        )
+        .execute(&mut *conn)
+        .await?;
+        sqlx::query!(
+            "DELETE FROM flow_conversation c
+              WHERE c.id = ANY($1)
+                AND NOT EXISTS (
+                    SELECT 1 FROM flow_conversation_message m WHERE m.conversation_id = c.id
+                )",
+            &conversation_ids
+        )
+        .execute(&mut *conn)
+        .await?;
+    }
     sqlx::query!("DELETE FROM zombie_job_counter WHERE job_id = ANY($1)", ids)
         .execute(&mut *conn)
         .await?;

backend/windmill-common/src/lib.rs

@@ -37,6 +37,10 @@ pub mod bench;
 pub mod cache;
 pub mod client;
 pub mod data_metrics;
+pub mod datatable_roles;
+#[cfg(all(feature = "private", feature = "enterprise"))]
+mod datatable_roles_ee;
+pub mod datatable_roles_oss;
 pub mod db;
 #[cfg(all(feature = "enterprise", feature = "private"))]
 mod db_entra_ee;
@@ -54,6 +58,10 @@ pub mod ee_oss;
 pub mod email_ee;
 pub mod email_oss;
 pub mod error;
+pub mod external_instance_pg;
+#[cfg(all(feature = "private", feature = "enterprise"))]
+mod external_instance_pg_ee;
+pub mod external_instance_pg_oss;
 pub mod external_ip;
 #[cfg(feature = "private")]
 pub mod feature_usage_ee;
@@ -1020,6 +1028,20 @@ impl Future for TokioPgConnection {
     }
 }
 
+impl TokioPgConnection {
+    /// Drive the connection and hand back what the server sends outside of a query's response —
+    /// notices above all, which driving it as a future silently discards.
+    pub fn poll_message(
+        &mut self,
+        cx: &mut core::task::Context<'_>,
+    ) -> core::task::Poll<Option<Result<tokio_postgres::AsyncMessage, tokio_postgres::Error>>> {
+        match self {
+            TokioPgConnection::Tls(conn) => conn.poll_message(cx),
+            TokioPgConnection::NoTls(conn) => conn.poll_message(cx),
+        }
+    }
+}
+
 impl PgDatabase {
     /// The role the connection logs in as, whichever way it authenticates.
     pub fn login_name(&self) -> &str {
@@ -1514,6 +1536,41 @@ pub async fn drop_custom_instance_database(db: &DB, dbname: &str) -> error::Resu
     Ok(())
 }
 
+/// What `custom_instance_user` holds on an instance database.
+///
+/// `WITH GRANT OPTION` throughout: this is the connection every data table resolves to as `admin`,
+/// and it is the one that hands privileges to data table roles. Postgres refuses to let a role pass
+/// on a privilege it does not itself hold with grant option, so without these an admin could own
+/// the database and still be unable to grant `SELECT` on it to `analytics`.
+pub(crate) fn instance_db_grants(dbname: &str) -> String {
+    format!(
+        "GRANT CONNECT ON DATABASE \"{dbname}\" TO custom_instance_user WITH GRANT OPTION;
+         GRANT CREATE ON DATABASE \"{dbname}\" TO custom_instance_user WITH GRANT OPTION;
+         DO $$ BEGIN
+           IF EXISTS (SELECT 1 FROM pg_namespace WHERE nspname = 'public') THEN
+             GRANT USAGE ON SCHEMA public TO custom_instance_user WITH GRANT OPTION;
+             GRANT CREATE ON SCHEMA public TO custom_instance_user WITH GRANT OPTION;
+             ALTER DEFAULT PRIVILEGES IN SCHEMA public
+                 GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO custom_instance_user;
+           END IF;
+         END $$;"
+    )
+}
+
+/// Re-apply [`instance_db_grants`] to an instance database provisioned before data table roles
+/// existed, whose grants carry no grant option. Connects as the instance's own Postgres user —
+/// the database and `public` schema owner — since only it can hand out an option it holds.
+///
+/// Authorization: reaches an instance database with the server's own credentials and checks
+/// nothing. Callers MUST have authorized administration of `dbname` — superadmin, or an admin of
+/// the workspace governing a data table on it.
+pub async fn ensure_instance_db_grant_options_unchecked(
+    db: &DB,
+    dbname: &str,
+) -> error::Result<()> {
+    crate::datatable_roles_oss::ensure_instance_db_grant_options_unchecked(db, dbname).await
+}
+
 /// Create a custom instance database: CREATE DATABASE, grant permissions, register in global_settings.
 /// The `tag` is stored in global_settings metadata (e.g. "datatable" or "ducklake").
 pub async fn create_custom_instance_database(
@@ -1553,17 +1610,7 @@ pub async fn create_custom_instance_database(
     let (client, connection) = new_pg_creds.connect(Some(db)).await?;
     let join_handle = tokio::spawn(async move { connection.await });
 
-    if let Err(e) = client
-        .batch_execute(&format!(
-            "GRANT CONNECT ON DATABASE \"{dbname}\" TO custom_instance_user;
-             GRANT USAGE ON SCHEMA public TO custom_instance_user;
-             GRANT CREATE ON SCHEMA public TO custom_instance_user;
-             GRANT CREATE ON DATABASE \"{dbname}\" TO custom_instance_user;
-             ALTER DEFAULT PRIVILEGES IN SCHEMA public
-                 GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO custom_instance_user;"
-        ))
-        .await
-    {
+    if let Err(e) = client.batch_execute(&instance_db_grants(dbname)).await {
         tracing::warn!(
             "Failed to grant permissions on '{}': {}. Continuing.",
             dbname,
@@ -1592,6 +1639,13 @@ pub async fn create_custom_instance_database(
     .execute(db)
     .await?;
 
+    // A data table role can only reach a database it may CONNECT to, and PUBLIC's default CONNECT
+    // would otherwise let every role in regardless of what this instance defines. Best-effort: a
+    // failure here leaves the database usable as `admin`, and the next role change repairs it.
+    if let Err(e) = crate::datatable_roles::converge_connect_grants(db, dbname).await {
+        tracing::warn!("Could not set CONNECT grants on instance database '{dbname}': {e}");
+    }
+
     tracing::info!("Created custom instance database '{}'", dbname);
     Ok(())
 }

backend/windmill-common/src/query_builders.rs

@@ -329,6 +329,7 @@ pub fn try_expand_internal_db_query(
         "ALTER_TABLE" => expand_alter_table(json_str, db_type).map(ExpandedQuery::sql),
         "CREATE_SCHEMA" => expand_create_schema(json_str, db_type).map(ExpandedQuery::sql),
         "DROP_SCHEMA" => expand_drop_schema(json_str, db_type).map(ExpandedQuery::sql),
+        "RENAME_SCHEMA" => expand_rename_schema(json_str, db_type).map(ExpandedQuery::sql),
         // Metadata queries
         "LOAD_TABLE_METADATA" => expand_load_table_metadata(json_str, db_type),
         "FOREIGN_KEYS" => expand_foreign_keys(json_str, db_type).map(ExpandedQuery::sql),
@@ -1716,6 +1717,13 @@ struct DropSchemaPayload {
     ducklake: Option<String>,
 }
 
+#[derive(Deserialize)]
+struct RenameSchemaPayload {
+    schema: String,
+    new_schema: String,
+    ducklake: Option<String>,
+}
+
 #[derive(Debug, Clone, Deserialize)]
 struct TableEditorColumn {
     name: String,
@@ -2004,6 +2012,23 @@ fn expand_drop_schema(json_str: &str, db_type: DbType) -> Result<String, String>
     Ok(maybe_wrap_ducklake(query, p.ducklake.as_deref()))
 }
 
+fn expand_rename_schema(json_str: &str, db_type: DbType) -> Result<String, String> {
+    let p: RenameSchemaPayload = serde_json::from_str(json_str)
+        .map_err(|e| format!("Invalid RENAME_SCHEMA payload: {}", e))?;
+    if !matches!(db_type, DbType::Postgresql | DbType::Snowflake) || p.ducklake.is_some() {
+        return Err(format!(
+            "Renaming a schema is not supported on {:?}",
+            db_type
+        ));
+    }
+    let query = format!(
+        "ALTER SCHEMA {} RENAME TO {};",
+        qi(&p.schema, db_type),
+        qi(&p.new_schema, db_type)
+    );
+    Ok(query)
+}
+
 fn expand_create_table(json_str: &str, db_type: DbType) -> Result<String, String> {
     let p: CreateTablePayload = serde_json::from_str(json_str)
         .map_err(|e| format!("Invalid CREATE_TABLE payload: {}", e))?;
@@ -2598,7 +2623,9 @@ WHERE table_catalog = current_database()",
                 )
             } else {
                 (
-                    "\nWHERE c.relkind = 'r' AND a.attnum > 0 AND NOT a.attisdropped\n    AND ns.nspname != 'pg_catalog' AND ns.nspname != 'information_schema'".to_string(),
+                    // pg_catalog is readable by everyone: without the privilege check this lists
+                    // tables of schemas the connection's role cannot even enter.
+                    "\nWHERE c.relkind = 'r' AND a.attnum > 0 AND NOT a.attisdropped\n    AND ns.nspname != 'pg_catalog' AND ns.nspname != 'information_schema'\n    AND has_schema_privilege(ns.oid, 'USAGE')".to_string(),
                     ",\n    ns.nspname AS schema_name,\n    c.relname AS table_name".to_string(),
                     "\nJOIN pg_catalog.pg_class c ON a.attrelid = c.oid\nJOIN pg_catalog.pg_namespace ns ON c.relnamespace = ns.oid".to_string(),
                     "ns.nspname, c.relname, a.attnum".to_string(),
@@ -4101,6 +4128,13 @@ mod tests {
         assert_eq!(sql, "DROP SCHEMA \"old_schema\" CASCADE;");
     }
 
+    #[test]
+    fn test_expand_rename_schema() {
+        let marker = r#"-- WM_INTERNAL_DB_RENAME_SCHEMA {"schema":"old","new_schema":"new"}"#;
+        let sql = expand_code(marker, &ScriptLang::Postgresql);
+        assert_eq!(sql, "ALTER SCHEMA \"old\" RENAME TO \"new\";");
+    }
+
     #[test]
     fn test_expand_create_schema_with_ducklake() {
         let marker = r#"-- WM_INTERNAL_DB_CREATE_SCHEMA {"schema":"s","ducklake":"lake"}"#;
@@ -4468,6 +4502,7 @@ mod tests {
         assert!(sql.contains("schema_name"));
         assert!(sql.contains("table_name"));
         assert!(sql.contains("c.relkind = 'r'"));
+        assert!(sql.contains("has_schema_privilege(ns.oid, 'USAGE')"));
     }
 
     #[test]

backend/windmill-common/src/worker.rs

@@ -1082,6 +1082,83 @@ pub struct SqlAnnotations {
     pub raw_output: bool,
 }
 
+impl SqlAnnotations {
+    /// The data table role a query declares as `-- role <name>`, if any. Only meaningful against a
+    /// `datatable://` database that is under roles; absent means the data table's default role.
+    ///
+    /// Hand-written rather than derived because the value matters, not just the presence, and
+    /// because the executor needs it before it knows the connection is a data table at all. Like
+    /// every annotation it lives in the leading comment block.
+    ///
+    /// A leading comment whose first word is `role` is an annotation *attempt*, and a malformed
+    /// one is an error. The alternative — ignoring what does not parse — resolves the query to the
+    /// data table's default role instead, so a typo silently runs it under a login the author did
+    /// not choose, which is the opposite of what naming a role is for. Only callers that already
+    /// know the target is a `datatable://` reference ever run this, so ordinary SQL keeps its
+    /// comments.
+    pub fn datatable_role(code: &str) -> error::Result<Option<String>> {
+        for line in code.lines() {
+            let line = line.trim();
+            if line.is_empty() {
+                continue;
+            }
+            if !line.starts_with("--") {
+                break;
+            }
+            // The keyword may be followed by whitespace, `:` or `=` — `role x`, `role: x`,
+            // `role=x`, `Role = x` all open an attempt, while `rolexyz` does not. Each accepted
+            // separator is one spelling that would otherwise take the `continue` below and run the
+            // query as the data table's default role, which is the silence this exists to remove.
+            let body = line[2..].trim_start();
+            let Some(after) = body
+                .get(..4)
+                .filter(|kw| kw.eq_ignore_ascii_case("role"))
+                .map(|_| &body[4..])
+            else {
+                continue;
+            };
+            if !after.is_empty()
+                && !after.starts_with(char::is_whitespace)
+                && !after.starts_with([':', '='])
+            {
+                continue;
+            }
+
+            // Past this point the line is an attempt to name a role, so a malformed one is an
+            // error rather than a miss. Falling through would run the query as the data table's
+            // default role — quietly, and under a login the author did not choose.
+            let after = after.trim_start();
+            let after = after.strip_prefix([':', '=']).unwrap_or(after);
+            let mut tokens = after.split_whitespace();
+            let role = tokens
+                .next()
+                .map(|role| role.strip_suffix(';').unwrap_or(role));
+            let rest = tokens.next();
+            match (role, rest) {
+                (Some(role), None)
+                    if !role.is_empty()
+                        && role.len() <= 63
+                        && role
+                            .chars()
+                            .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-') =>
+                {
+                    return Ok(Some(role.to_string()));
+                }
+                _ => {
+                    return Err(error::Error::BadRequest(format!(
+                        "Malformed data table role annotation: `{line}`. Write it as \
+                         `-- role <name>` on a line of its own, where <name> is letters, digits, \
+                         '_' or '-'. A comment in the leading block that starts with the word \
+                         'role' is read as this annotation; move it below the first statement if \
+                         it is prose."
+                    )));
+                }
+            }
+        }
+        Ok(None)
+    }
+}
+
 #[annotations("#")]
 pub struct BashAnnotations {
     pub docker: bool,
@@ -2653,6 +2730,56 @@ mod tests {
     use super::*;
     use std::collections::HashMap;
 
+    #[test]
+    fn datatable_role_is_read_from_the_leading_comment_block() {
+        let role = |code| SqlAnnotations::datatable_role(code);
+        assert_eq!(
+            role("-- role analytics\nSELECT 1").unwrap(),
+            Some("analytics".to_string())
+        );
+        // Blank lines and other annotations before it are fine.
+        assert_eq!(
+            role("\n-- prepare\n-- role read_only\nSELECT 1").unwrap(),
+            Some("read_only".to_string())
+        );
+        // Past the first statement it is an ordinary comment, not an annotation.
+        assert_eq!(role("SELECT 1;\n-- role analytics").unwrap(), None);
+        assert_eq!(role("SELECT 1").unwrap(), None);
+
+        // Unambiguous intent is honoured: the keyword matches case-insensitively, a trailing
+        // semicolon is a habit carried over from SQL rather than a different role, and the colon
+        // spelling is the one most likely to be typed.
+        for accepted in [
+            "-- Role operator\nSELECT 1",
+            "-- role operator;\nSELECT 1",
+            "-- role: operator\nSELECT 1",
+            "-- role:operator\nSELECT 1",
+            "-- role=operator\nSELECT 1",
+            "-- Role = operator\nSELECT 1",
+        ] {
+            assert_eq!(
+                role(accepted).unwrap(),
+                Some("operator".to_string()),
+                "not honoured: {accepted}"
+            );
+        }
+
+        // Anything else opening with the word is refused rather than resolved to the default role:
+        // the whole point of naming one is to not run as something else.
+        for near_miss in [
+            "-- role operator -- why\nSELECT 1",
+            "-- role an;alytics\nSELECT 1",
+            "-- role\nSELECT 1",
+            "-- role:\nSELECT 1",
+            "-- role based access is handled below\nSELECT 1",
+        ] {
+            assert!(role(near_miss).is_err(), "silently ignored: {near_miss}");
+        }
+
+        // A word that merely starts with the keyword is not an attempt.
+        assert_eq!(role("-- rolebased notes\nSELECT 1").unwrap(), None);
+    }
+
     fn matcher(id: &str) -> WorkspaceMatcher {
         WorkspaceMatcher { id: id.to_string(), include_forks: false }
     }

backend/windmill-common/src/workspaces.rs

@@ -6,6 +6,7 @@ use serde::{Deserialize, Serialize};
 use strum::AsRefStr;
 
 use crate::{
+    datatable_roles::{ADMIN_DATATABLE_ROLE, CUSTOM_INSTANCE_USER},
     error::{self, to_anyhow, Error, Result},
     get_database_url,
     secret_backend::{get_secret_value, is_external_stored_value},
@@ -1292,9 +1293,17 @@ impl Default for DataTableForkBehavior {
     }
 }
 
-#[derive(Deserialize, Serialize, Debug)]
+#[derive(Deserialize, Serialize, Debug, Clone)]
 pub struct DataTable {
-    pub database: DataTableDatabase,
+    /// Set on a *terminal* entry — one that owns its database. Mutually exclusive with
+    /// [`DataTable::reference`]; [`validate_datatable_shape`] is the one place that enforces it.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub database: Option<DataTableDatabase>,
+    /// Set on a *pointer* entry — one that names another workspace's entry and owns nothing.
+    /// A keep-original fork gets one of these instead of a copy of the parent's entry, so there is
+    /// nothing local for a fork admin to widen.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub reference: Option<DataTableReference>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub forked_from: Option<DataTableForkedFrom>,
     /// Whether the SQL-migrations feature is opted in for this data table.
@@ -1302,22 +1311,85 @@ pub struct DataTable {
     /// when migrations already exist (see `datatable_migrations_enabled`).
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub migrations_enabled: Option<bool>,
+    /// Who may connect as which role. Absent = unpermissioned: every caller connects as `admin`,
+    /// which is how data tables behaved before roles existed. Only meaningful on a terminal entry;
+    /// a pointer is governed by what it points at.
+    ///
+    /// Never leaves the instance: stripped from the workspace export and ignored on import, since
+    /// tenants are workspace-scoped names and syncing them would make repo write access a second
+    /// door onto the access decision.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub permissions: Option<DataTablePermissions>,
 }
 
-#[derive(Deserialize, Serialize, Debug)]
+/// A pointer at another workspace's data table entry.
+#[derive(Deserialize, Serialize, Debug, Clone, PartialEq)]
+pub struct DataTableReference {
+    pub workspace_id: String,
+    pub datatable: String,
+}
+
+/// The access decision for one data table: which role a caller gets, and who may ask for each.
+#[derive(Deserialize, Serialize, Debug, Clone, Default)]
+pub struct DataTablePermissions {
+    /// A role id from the instance catalog, or `admin`. Absent = `admin`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub default_role: Option<String>,
+    /// Keyed by instance role id, plus the reserved `admin` key. A role absent from this map
+    /// cannot be used on this data table at all, whatever the instance catalog says.
+    #[serde(default)]
+    pub roles: std::collections::BTreeMap<String, DataTableRoleTenants>,
+}
+
+impl DataTablePermissions {
+    pub fn default_role(&self) -> &str {
+        self.default_role.as_deref().unwrap_or(ADMIN_DATATABLE_ROLE)
+    }
+}
+
+#[derive(Deserialize, Serialize, Debug, Clone, Default)]
+pub struct DataTableRoleTenants {
+    /// `u/<user>`, `g/<group>`, `f/<folder>`, or `*` for every member.
+    #[serde(default)]
+    pub tenants: Vec<String>,
+}
+
+/// Every member of the governing workspace.
+pub const DATATABLE_TENANT_WILDCARD: &str = "*";
+
+/// How deep a chain of pointer entries may go before it is called a loop. Data tables are not
+/// expected to chain at all — a fork points at its parent — so this only has to be generous
+/// enough to survive a fork of a fork.
+const DATATABLE_REFERENCE_MAX_DEPTH: usize = 20;
+
+/// Exactly one of `database` and `reference` must be set. Called wherever an entry is persisted,
+/// so nothing downstream has to handle an entry that is both or neither.
+pub fn validate_datatable_shape(name: &str, dt: &DataTable) -> Result<()> {
+    match (&dt.database, &dt.reference) {
+        (Some(_), None) | (None, Some(_)) => Ok(()),
+        (Some(_), Some(_)) => Err(Error::BadRequest(format!(
+            "Data table '{name}' both owns a database and points at another one"
+        ))),
+        (None, None) => Err(Error::BadRequest(format!(
+            "Data table '{name}' names neither a database nor another data table"
+        ))),
+    }
+}
+
+#[derive(Deserialize, Serialize, Debug, Clone)]
 pub struct DataTableForkedFrom {
     /// Schema snapshot at fork time
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub schema: Option<serde_json::Value>,
 }
 
-#[derive(Deserialize, Serialize, Debug)]
+#[derive(Deserialize, Serialize, Debug, Clone)]
 pub struct DataTableDatabase {
     pub resource_type: DataTableCatalogResourceType,
     pub resource_path: String,
 }
 
-#[derive(Deserialize, Serialize, Debug, PartialEq)]
+#[derive(Deserialize, Serialize, Debug, PartialEq, Clone, Copy)]
 #[serde(rename_all = "lowercase")]
 #[derive(AsRefStr)]
 #[strum(serialize_all = "lowercase")]
@@ -1353,37 +1425,13 @@ fn datatable_not_found_error(name: &str, datatables: Option<&serde_json::Value>)
     ))
 }
 
-pub async fn get_datatable_resource_from_db_unchecked(
-    db: &DB,
-    w_id: &str,
-    name: &str,
-) -> Result<serde_json::Value> {
-    get_datatable_resource_inner(db, w_id, name, false).await
-}
-
-/// Same as [`get_datatable_resource_from_db_unchecked`] but for postgres trigger
-/// connections: custom-instance datatables resolve to
-/// `custom_instance_replication_user` rather than `custom_instance_user`. BYO-postgres
-/// datatables resolve to the user's own resource unchanged; configuring it for
-/// replication there is the user's responsibility.
+/// Read one workspace's data table entry, without following a pointer.
 ///
-/// Authorization: like its `_unchecked` sibling, returns resolved connection
-/// credentials and performs no authorization — callers MUST have already authorized
-/// access to the datatable (e.g. the trigger's own create-time check).
-pub async fn get_datatable_replication_resource_from_db_unchecked(
-    db: &DB,
-    w_id: &str,
-    name: &str,
-) -> Result<serde_json::Value> {
-    get_datatable_resource_inner(db, w_id, name, true).await
-}
-
-async fn get_datatable_resource_inner(
-    db: &DB,
-    w_id: &str,
-    name: &str,
-    replication: bool,
-) -> Result<serde_json::Value> {
+/// Disclosure: this is the primitive [`resolve_governing_datatable`] calls on every path, so it is
+/// deliberately open to anything that has to resolve a data table, including for a workspace the
+/// caller does not belong to. What it returns is not: callers MUST NOT put `permissions` into a
+/// response, an export or a log — it names the governing workspace's users, groups and folders.
+pub async fn read_datatable_entry(db: &DB, w_id: &str, name: &str) -> Result<DataTable> {
     let datatables = sqlx::query_scalar!(
         r#"
             SELECT ws.datatable->'datatables' AS datatables
@@ -1401,37 +1449,508 @@ async fn get_datatable_resource_inner(
         .and_then(|d| d.get(name))
         .filter(|v| !v.is_null())
         .ok_or_else(|| datatable_not_found_error(name, datatables.as_ref()))?;
-    let datatable = serde_json::from_value::<DataTable>(datatable.clone())?;
+    Ok(serde_json::from_value::<DataTable>(datatable.clone())?)
+}
 
-    let db_resource = if datatable.database.resource_type == DataTableCatalogResourceType::Instance
-    {
+/// The terminal entry a reference chain lands on: the workspace that governs the data table, the
+/// entry name there, and the entry itself. A terminal entry resolves to itself.
+///
+/// Every decision downstream — which database to connect to, whose `permissions` apply, whose
+/// members tenants are evaluated against, who may administer it — is taken on this, never on the
+/// entry the caller named.
+///
+/// Authorization: resolving deliberately crosses into the governing workspace, so it answers for a
+/// workspace the caller may not belong to and checks nothing itself. It is the input to the
+/// checks, not one of them: callers MUST pass what it returns to
+/// [`can_use_datatable_role_in_governing_workspace`] or [`ensure_datatable_admin_access`] before
+/// acting on it, and MUST NOT return its `permissions` or `workspace_id` to a caller from
+/// elsewhere without gating on the answer.
+pub struct GoverningDatatable {
+    pub workspace_id: String,
+    pub name: String,
+    pub datatable: DataTable,
+}
+
+impl GoverningDatatable {
+    /// Backed by the Windmill instance's own Postgres, which is the only substrate data table
+    /// roles apply to.
+    pub fn is_instance(&self) -> bool {
+        self.datatable
+            .database
+            .as_ref()
+            .is_some_and(|d| d.resource_type == DataTableCatalogResourceType::Instance)
+    }
+}
+
+pub async fn resolve_governing_datatable(
+    db: &DB,
+    w_id: &str,
+    name: &str,
+) -> Result<GoverningDatatable> {
+    let mut workspace_id = w_id.to_string();
+    let mut name = name.to_string();
+    let mut hops = 0;
+    for _ in 0..DATATABLE_REFERENCE_MAX_DEPTH {
+        let datatable = read_datatable_entry(db, &workspace_id, &name)
+            .await
+            .map_err(|e| {
+                if hops == 0 {
+                    e
+                } else {
+                    // A pointer outlives the workspace it names: deleting one only nulls the fork
+                    // lineage, it does not sweep the entries that pointed at it. Say which one is
+                    // gone rather than reporting a data table this workspace never had.
+                    Error::NotFound(format!(
+                        "Data table '{name}' of workspace '{workspace_id}' governs this one and no \
+                         longer exists. A superadmin can point this data table somewhere else."
+                    ))
+                }
+            })?;
+        hops += 1;
+        validate_datatable_shape(&name, &datatable)?;
+        match &datatable.reference {
+            None => return Ok(GoverningDatatable { workspace_id, name, datatable }),
+            Some(reference) => {
+                workspace_id = reference.workspace_id.clone();
+                name = reference.datatable.clone();
+            }
+        }
+    }
+    Err(Error::BadRequest(format!(
+        "Data table '{name}' points at another data table through more than \
+         {DATATABLE_REFERENCE_MAX_DEPTH} hops; the chain is likely a loop"
+    )))
+}
+
+/// Build the `admin` connection for a governing entry: `custom_instance_user` for an instance
+/// database, the user's own resource for a BYO-postgres one.
+async fn resolve_datatable_connection_unchecked(
+    db: &DB,
+    governing: &GoverningDatatable,
+    replication: bool,
+) -> Result<serde_json::Value> {
+    let database = governing
+        .datatable
+        .database
+        .as_ref()
+        .expect("a governing entry owns a database");
+    if database.resource_type == DataTableCatalogResourceType::Instance {
         let mut pg_creds = PgDatabase::parse_uri(&get_database_url().await?.as_str().await)?;
-        pg_creds.dbname = datatable.database.resource_path.clone();
+        pg_creds.dbname = database.resource_path.clone();
         if replication {
             pg_creds.user = Some("custom_instance_replication_user".to_string());
             pg_creds.password = Some(get_custom_pg_instance_replication_password(&db).await?);
         } else {
-            pg_creds.user = Some("custom_instance_user".to_string());
+            pg_creds.user = Some(CUSTOM_INSTANCE_USER.to_string());
             pg_creds.password = Some(get_custom_pg_instance_password(&db).await?);
         }
         serde_json::to_value(&pg_creds)
-            .map_err(|e| Error::internal_err(format!("Error serializing pg creds: {}", e)))?
+            .map_err(|e| Error::internal_err(format!("Error serializing pg creds: {}", e)))
     } else {
         // Name the data table too: the caller asked for one by name, and a bare
         // "resource f/x/y does not exist" leaves them to work out which one points at it.
         transform_json_unchecked(
-            &serde_json::Value::String(format!("$res:{}", datatable.database.resource_path)),
-            w_id,
+            &serde_json::Value::String(format!("$res:{}", database.resource_path)),
+            &governing.workspace_id,
             db,
         )
         .await
         .map_err(|e| match e {
-            Error::NotFound(m) => Error::NotFound(format!("data table {name}: {m}")),
+            Error::NotFound(m) => Error::NotFound(format!("data table {}: {m}", governing.name)),
             e => e,
-        })?
+        })
+    }
+}
+
+/// Resolve a data table to connection credentials **without authorizing anything**: always the
+/// `admin` connection.
+///
+/// Authorization: callers MUST have authorized access already. Anything that acts for a user or a
+/// job wants [`get_datatable_resource_from_db`] instead.
+pub async fn get_datatable_resource_from_db_unchecked(
+    db: &DB,
+    w_id: &str,
+    name: &str,
+) -> Result<serde_json::Value> {
+    let governing = resolve_governing_datatable(db, w_id, name).await?;
+    resolve_datatable_connection_unchecked(db, &governing, false).await
+}
+
+/// Same as [`get_datatable_resource_from_db_unchecked`] but for postgres trigger
+/// connections: custom-instance datatables resolve to
+/// `custom_instance_replication_user` rather than `custom_instance_user`. BYO-postgres
+/// datatables resolve to the user's own resource unchanged; configuring it for
+/// replication there is the user's responsibility.
+///
+/// Authorization: a replication connection reads every row whatever the roles grant, so no role or
+/// admin check makes it safe. Callers MUST refuse a data table under roles outright — the Postgres
+/// trigger crate's `ensure_not_under_roles` — and turning roles on is refused while one streams.
+pub async fn get_datatable_replication_resource_from_db_unchecked(
+    db: &DB,
+    w_id: &str,
+    name: &str,
+) -> Result<serde_json::Value> {
+    let governing = resolve_governing_datatable(db, w_id, name).await?;
+    resolve_datatable_connection_unchecked(db, &governing, true).await
+}
+
+/// The identity a resolution is made for. `Unchecked` is for callers that authorized already;
+/// everything else is checked against the governing entry's tenants.
+pub enum DatatableAccess<'a> {
+    /// Reaches every role. For callers that already authorized, or that have no user at all.
+    Unchecked,
+    Authed(crate::db::AuthedRef<'a>),
+    /// A job's owner, without reading the job row — only fetched if the data table turns out to
+    /// be permissioned.
+    PermissionedAs {
+        permissioned_as: &'a str,
+        email: &'a str,
+    },
+    /// A job identified by id; its owner is read from `v2_job`. For agent workers and anything
+    /// else that authenticates as infrastructure rather than as the job's user.
+    Job(uuid::Uuid),
+    /// No identity established. Unpermissioned data tables resolve as before; permissioned ones
+    /// are refused, so a caller predating this feature fails closed.
+    NoIdentity,
+}
+
+/// Does one tenant list cover this identity? Admins of the governing workspace pass everything —
+/// they can edit the tenant lists anyway, so refusing them would only be theatre.
+pub fn can_use_datatable_role(
+    tenants: &DataTableRoleTenants,
+    authed: &crate::db::AuthedRef<'_>,
+) -> bool {
+    crate::datatable_roles_oss::can_use_datatable_role(tenants, authed)
+}
+
+/// Evaluate a tenant list **as a member of the governing workspace**, whoever is calling.
+///
+/// A caller reaching a data table through a pointer is a member of some other workspace, and being
+/// its admin means nothing here — that is the whole point of the pointer. They are looked up in
+/// the governing workspace by email and evaluated there, or refused when they are not a member.
+/// A `g/` or `f/` permissioned-as from a foreign workspace is refused outright: those names are
+/// defined per workspace and mean nothing outside the one that defined them.
+pub async fn can_use_datatable_role_in_governing_workspace(
+    db: &DB,
+    governing_w_id: &str,
+    w_id: &str,
+    tenants: &DataTableRoleTenants,
+    access: &DatatableAccess<'_>,
+) -> Result<bool> {
+    crate::datatable_roles_oss::can_use_datatable_role_in_governing_workspace(
+        db,
+        governing_w_id,
+        w_id,
+        tenants,
+        access,
+    )
+    .await
+}
+
+/// Resolve a data table to connection credentials for one identity.
+///
+/// This is the chokepoint: everything that opens a connection to a data table on someone's behalf
+/// goes through it. `role` is the role name the caller asked for — the `-- role` annotation, the
+/// `?role=` on a `datatable://` reference, or `None` for the data table's default.
+///
+/// The resolved role **logs in as itself**. Never `SET ROLE`: a script could `RESET ROLE` its way
+/// back to admin.
+pub async fn get_datatable_resource_from_db(
+    db: &DB,
+    w_id: &str,
+    name: &str,
+    role: Option<&str>,
+    access: DatatableAccess<'_>,
+) -> Result<serde_json::Value> {
+    let governing = resolve_governing_datatable(db, w_id, name).await?;
+    let db_resource = resolve_datatable_connection_unchecked(db, &governing, false).await?;
+    // Not under roles and asked for none, or for `admin` by name: the `admin` connection, as before
+    // roles existed, in every edition. Anything else is a role decision. Every migration names
+    // `admin` explicitly, so an edition without roles must not treat that as one.
+    if governing.datatable.permissions.is_none() && role.is_none_or(|r| r == ADMIN_DATATABLE_ROLE) {
+        return Ok(db_resource);
+    }
+    crate::datatable_roles_oss::resolve_datatable_role_connection(
+        db,
+        w_id,
+        name,
+        &governing,
+        db_resource,
+        role,
+        access,
+    )
+    .await
+}
+
+/// Would the chokepoint accept this identity connecting as this role? Answers without resolving
+/// credentials, for callers that want to refuse early and say which thing was refused.
+///
+/// Not the security boundary — [`get_datatable_resource_from_db`] re-checks when it actually opens
+/// the connection. This is what turns "permission denied for table x" into a message naming the
+/// migration and the role.
+pub async fn ensure_can_use_datatable_role(
+    db: &DB,
+    w_id: &str,
+    name: &str,
+    role: Option<&str>,
+    access: &DatatableAccess<'_>,
+    context: &str,
+) -> Result<()> {
+    crate::datatable_roles_oss::ensure_can_use_datatable_role(db, w_id, name, role, access, context)
+        .await
+}
+
+/// Gate the operations that see the whole database whatever the roles grant: a migration that
+/// declares no role, exports, and editing the permissions themselves. Not replication, which a
+/// data table under roles refuses whoever asks (see `ensure_not_under_roles`). Passing
+/// means the caller could have connected as `admin` anyway.
+pub async fn ensure_datatable_admin_access(
+    db: &DB,
+    w_id: &str,
+    name: &str,
+    access: &DatatableAccess<'_>,
+) -> Result<()> {
+    crate::datatable_roles_oss::ensure_datatable_admin_access(db, w_id, name, access).await
+}
+
+/// Rewrite the `permissions` of every data table entry of one workspace, in the caller's
+/// transaction. `change` reports whether it touched anything; the row is only written when
+/// something did.
+///
+/// Authorization: writes an access decision for any workspace named, with an arbitrary mutation,
+/// and checks nothing. It exists for the cascades below — the transaction that frees or renames a
+/// principal — so callers MUST be the operation that made the principal change, and MUST run in
+/// its transaction. Anything editing a decision on purpose belongs in the permissions endpoint,
+/// which is gated on the workspace that governs the data table.
+///
+/// The tenant lists name principals of this workspace, so anything that frees or renames one has
+/// to come through here in the same transaction that frees it — otherwise a `u/alice` reused by a
+/// later account silently inherits her access.
+pub async fn update_datatable_permissions_in_workspace<F>(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    w_id: &str,
+    change: F,
+) -> Result<()>
+where
+    F: Fn(&mut DataTablePermissions) -> bool,
+{
+    let Some(mut settings) = sqlx::query_scalar!(
+        "SELECT datatable FROM workspace_settings WHERE workspace_id = $1 FOR UPDATE",
+        w_id
+    )
+    .fetch_optional(&mut **tx)
+    .await?
+    .flatten() else {
+        return Ok(());
     };
 
-    Ok(db_resource)
+    let Some(datatables) = settings
+        .get_mut("datatables")
+        .and_then(|d| d.as_object_mut())
+    else {
+        return Ok(());
+    };
+
+    let mut touched = false;
+    for entry in datatables.values_mut() {
+        let Some(permissions) = entry.get("permissions").filter(|p| !p.is_null()) else {
+            continue;
+        };
+        let Ok(mut permissions) =
+            serde_json::from_value::<DataTablePermissions>(permissions.clone())
+        else {
+            continue;
+        };
+        if change(&mut permissions) {
+            entry["permissions"] = serde_json::to_value(&permissions)
+                .map_err(|e| Error::internal_err(format!("serializing permissions: {e}")))?;
+            touched = true;
+        }
+    }
+
+    if touched {
+        sqlx::query!(
+            "UPDATE workspace_settings SET datatable = $1 WHERE workspace_id = $2",
+            settings,
+            w_id
+        )
+        .execute(&mut **tx)
+        .await?;
+    }
+    Ok(())
+}
+
+/// Drop a freed principal (`u/alice`, `g/analysts`, `f/finance`) from every tenant list of one
+/// workspace. Same contract as [`update_datatable_permissions_in_workspace`]: for the transaction
+/// that frees the principal, not for editing a decision.
+pub async fn remove_datatable_tenant_in_workspace(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    w_id: &str,
+    tenant: &str,
+) -> Result<()> {
+    update_datatable_permissions_in_workspace(tx, w_id, |permissions| {
+        let mut touched = false;
+        for role in permissions.roles.values_mut() {
+            let before = role.tenants.len();
+            role.tenants.retain(|t| t != tenant);
+            touched |= role.tenants.len() != before;
+        }
+        touched
+    })
+    .await
+}
+
+/// Follow a renamed principal through every tenant list of one workspace. Same contract as
+/// [`update_datatable_permissions_in_workspace`].
+pub async fn rename_datatable_tenant_in_workspace(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    w_id: &str,
+    old: &str,
+    new: &str,
+) -> Result<()> {
+    update_datatable_permissions_in_workspace(tx, w_id, |permissions| {
+        let mut touched = false;
+        for role in permissions.roles.values_mut() {
+            let mut role_touched = false;
+            for tenant in role.tenants.iter_mut() {
+                if tenant == old {
+                    *tenant = new.to_string();
+                    role_touched = true;
+                }
+            }
+            if role_touched {
+                // The rename can collide with a name already in the list, and the two need not be
+                // adjacent — `Vec::dedup` only collapses neighbours, so it would leave the pair.
+                let mut seen = std::collections::HashSet::new();
+                role.tenants.retain(|t| seen.insert(t.clone()));
+                touched = true;
+            }
+        }
+        touched
+    })
+    .await
+}
+
+/// Strip a deleted instance role from every workspace that had tenanted it, so nothing is left
+/// naming a role that no longer exists.
+///
+/// Authorization: reaches every workspace on the instance. Callers MUST be the superadmin path
+/// dropping the role from the cluster — it exists to follow that, not to edit tenants.
+///
+/// Takes that path's transaction rather than opening its own: run afterwards, a failure part-way
+/// leaves the catalog row already gone, so the retry answers `NotFound` while some workspaces
+/// still name a role nothing can connect as. In the transaction, the cluster drop, the catalog row
+/// and every tenant list commit together or not at all. A data table whose default role was the deleted one falls
+/// back to `admin` — the one role that is always present.
+pub async fn forget_datatable_role_everywhere(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    role_id: &str,
+) -> Result<()> {
+    crate::datatable_roles_oss::forget_datatable_role_everywhere(tx, role_id).await
+}
+
+/// Drop the `permissions` block from a `workspace_settings.datatable` value before it leaves the
+/// server.
+///
+/// Who may connect as which role is an access decision, not configuration, and its tenants name
+/// principals of one workspace — `g/analysts` in dev is a different group from `g/analysts` in
+/// prod. Shipping it would both mean nothing at the far end and turn a settings push into a way to
+/// widen access, so the decision stays where it was made. [`DataTable`] deserializes fine without
+/// it, and the settings-editing endpoint carries the stored block across untouched.
+pub fn strip_datatable_permissions(
+    datatable: Option<serde_json::Value>,
+) -> Option<serde_json::Value> {
+    let mut datatable = datatable?;
+    if let Some(entries) = datatable
+        .get_mut("datatables")
+        .and_then(|d| d.as_object_mut())
+    {
+        for entry in entries.values_mut() {
+            if let Some(entry) = entry.as_object_mut() {
+                entry.remove("permissions");
+            }
+        }
+    }
+    Some(datatable)
+}
+
+/// As [`parse_datatable_ref`], except that an entry whose stored name itself contains `?` — which
+/// names could before they were restricted — resolves by that exact name, without a role. It is
+/// looked up first, so `sales?role=x` never reaches a different entry than the one stored so.
+/// When `sales` is stored too, the reference means either one, and is refused rather than
+/// resolved to whichever is looked up first.
+pub async fn parse_datatable_ref_for(
+    db: &DB,
+    w_id: &str,
+    reference: &str,
+) -> Result<(String, Option<String>)> {
+    if reference.contains('?') {
+        let role_target = parse_datatable_ref(reference)
+            .ok()
+            .and_then(|(name, role)| role.map(|_| name));
+        let (exists, target_exists) = sqlx::query_as::<_, (Option<bool>, Option<bool>)>(
+            "SELECT (datatable->'datatables') ? $2, (datatable->'datatables') ? $3
+             FROM workspace_settings WHERE workspace_id = $1",
+        )
+        .bind(w_id)
+        .bind(reference)
+        .bind(role_target)
+        .fetch_optional(db)
+        .await?
+        .unwrap_or((None, None));
+        if exists.unwrap_or(false) {
+            if let (Some(name), Some(true)) = (role_target, target_exists) {
+                return Err(Error::BadRequest(format!(
+                    "Data table reference '{reference}' names both the data table '{reference}' \
+                     and a role on the data table '{name}'. Rename '{reference}' to use either."
+                )));
+            }
+            return Ok((reference.to_string(), None));
+        }
+    }
+    let (name, role) = parse_datatable_ref(reference)?;
+    Ok((name.to_string(), role.map(str::to_string)))
+}
+
+/// Split a `datatable://` reference into its name and the role its query string names.
+///
+/// A query string that does not parse is an error rather than an absent role. Falling back would
+/// resolve the reference to the data table's default role, so `?Role=analytics` or a mistyped
+/// `?role=` would quietly connect as something the caller did not ask for — the same trap as a
+/// malformed `-- role` annotation, and `role` is the only parameter a reference takes.
+pub fn parse_datatable_ref(reference: &str) -> Result<(&str, Option<&str>)> {
+    let (name, query) = reference.split_once('?').unwrap_or((reference, ""));
+    let mut role = None;
+    for param in query.split('&').filter(|p| !p.is_empty()) {
+        let (key, value) = param.split_once('=').unwrap_or((param, ""));
+        if !key.eq_ignore_ascii_case("role") {
+            return Err(Error::BadRequest(format!(
+                "Data table reference '{name}' carries an unknown parameter '{key}'. \
+                 The only one it takes is `?role=<name>`."
+            )));
+        }
+        if role.is_some() {
+            return Err(Error::BadRequest(format!(
+                "Data table reference '{name}' names a role more than once."
+            )));
+        }
+        if value.is_empty() || !is_datatable_role_name(value) {
+            return Err(Error::BadRequest(format!(
+                "Data table reference '{name}' has a malformed role '{value}'. Write it as \
+                 `?role=<name>`, where <name> is letters, digits, '_' or '-'."
+            )));
+        }
+        role = Some(value);
+    }
+    Ok((name, role))
+}
+
+fn is_datatable_role_name(role: &str) -> bool {
+    !role.is_empty()
+        && role.len() <= 63
+        && role
+            .chars()
+            .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
 }
 
 #[derive(Deserialize, Serialize, Debug)]
@@ -2642,6 +3161,130 @@ async fn transform_json_unchecked(
 mod tests {
     use super::*;
 
+    fn tenants(list: &[&str]) -> DataTableRoleTenants {
+        DataTableRoleTenants { tenants: list.iter().map(|t| t.to_string()).collect() }
+    }
+
+    #[cfg(all(feature = "private", feature = "enterprise"))]
+    #[test]
+    fn a_tenant_list_covers_users_groups_folders_and_the_wildcard() {
+        let groups = vec!["analysts".to_string()];
+        let folders = vec![("finance".to_string(), true, false)];
+        let scopes = None;
+        let token_prefix = None;
+        let is_admin = false;
+        let is_operator = false;
+        let authed = crate::db::AuthedRef {
+            email: "alice@windmill.dev",
+            username: "alice",
+            is_admin: &is_admin,
+            is_operator: &is_operator,
+            groups: &groups,
+            folders: &folders,
+            scopes: &scopes,
+            token_prefix: &token_prefix,
+        };
+
+        assert!(can_use_datatable_role(&tenants(&["u/alice"]), &authed));
+        assert!(can_use_datatable_role(&tenants(&["g/analysts"]), &authed));
+        assert!(can_use_datatable_role(&tenants(&["f/finance"]), &authed));
+        assert!(can_use_datatable_role(&tenants(&["*"]), &authed));
+        assert!(!can_use_datatable_role(&tenants(&[]), &authed));
+        assert!(!can_use_datatable_role(
+            &tenants(&["u/bob", "g/ops"]),
+            &authed
+        ));
+        // A bare name is not a principal: only the three prefixes and the wildcard match.
+        assert!(!can_use_datatable_role(&tenants(&["alice"]), &authed));
+
+        // An admin of the governing workspace reaches every role: they can edit the lists anyway.
+        let is_admin = true;
+        let admin = crate::db::AuthedRef { is_admin: &is_admin, ..authed };
+        assert!(can_use_datatable_role(&tenants(&[]), &admin));
+    }
+
+    #[cfg(not(all(feature = "private", feature = "enterprise")))]
+    #[test]
+    fn without_the_enterprise_edition_no_tenant_list_covers_anyone() {
+        let groups = vec![];
+        let folders = vec![];
+        let scopes = None;
+        let token_prefix = None;
+        let is_admin = true;
+        let is_operator = false;
+        let admin = crate::db::AuthedRef {
+            email: "alice@windmill.dev",
+            username: "alice",
+            is_admin: &is_admin,
+            is_operator: &is_operator,
+            groups: &groups,
+            folders: &folders,
+            scopes: &scopes,
+            token_prefix: &token_prefix,
+        };
+        assert!(!can_use_datatable_role(&tenants(&["*"]), &admin));
+        assert!(!can_use_datatable_role(&tenants(&["u/alice"]), &admin));
+    }
+
+    #[test]
+    fn a_datatable_ref_splits_off_its_role() {
+        assert_eq!(parse_datatable_ref("sales").unwrap(), ("sales", None));
+        assert_eq!(
+            parse_datatable_ref("sales?role=analytics").unwrap(),
+            ("sales", Some("analytics"))
+        );
+        // The key matches case-insensitively, the way the `-- role` annotation does.
+        assert_eq!(
+            parse_datatable_ref("sales?Role=analytics").unwrap(),
+            ("sales", Some("analytics"))
+        );
+
+        // A query string that does not parse is refused rather than read as "no role": resolving
+        // it to the data table's default would connect as a login the caller never asked for.
+        for malformed in [
+            "sales?role=",
+            "sales?role=an;alytics",
+            "sales?x=1&role=analytics",
+            "sales?role=a&role=b",
+        ] {
+            assert!(
+                parse_datatable_ref(malformed).is_err(),
+                "silently ignored: {malformed}"
+            );
+        }
+    }
+
+    #[test]
+    fn an_entry_owns_a_database_or_points_at_one_but_never_both() {
+        let terminal = DataTable {
+            database: Some(DataTableDatabase {
+                resource_type: DataTableCatalogResourceType::Instance,
+                resource_path: "dt_main".to_string(),
+            }),
+            reference: None,
+            forked_from: None,
+            migrations_enabled: None,
+            permissions: None,
+        };
+        assert!(validate_datatable_shape("main", &terminal).is_ok());
+
+        let pointer = DataTable {
+            database: None,
+            reference: Some(DataTableReference {
+                workspace_id: "prod".to_string(),
+                datatable: "main".to_string(),
+            }),
+            ..terminal.clone()
+        };
+        assert!(validate_datatable_shape("main", &pointer).is_ok());
+
+        let both = DataTable { database: terminal.database.clone(), ..pointer.clone() };
+        assert!(validate_datatable_shape("main", &both).is_err());
+
+        let neither = DataTable { database: None, reference: None, ..terminal.clone() };
+        assert!(validate_datatable_shape("main", &neither).is_err());
+    }
+
     #[test]
     fn test_parse_fork_branch() {
         // Generated fork (`wm-fork-abc`) and dev workspace (`staging`) forms.

backend/windmill-trigger-postgres/src/handler.rs

@@ -21,11 +21,12 @@ use windmill_common::{
 use windmill_git_sync::DeployedObject;
 
 use windmill_api_auth::{check_scopes, ApiAuthed};
-use windmill_trigger::{Trigger, TriggerCrud, TriggerData};
+use windmill_trigger::{Trigger, TriggerCrud, TriggerData, TriggerMode};
 
 use super::{
     check_if_valid_publication_for_postgres_version, create_logical_replication_slot,
-    create_pg_publication, drop_publication, generate_random_string, get_default_pg_connection,
+    create_pg_publication, drop_publication, ensure_not_under_roles, generate_random_string,
+    get_default_pg_connection,
     mapper::{Mapper, MappingInfo},
     PostgresConfig, PostgresConfigRequest, PostgresPublicationReplication, PostgresTrigger,
     PublicationData, Relations, Slot, SlotList, TableToTrack, TemplateScript, TestPostgresConfig,
@@ -64,6 +65,29 @@ impl TriggerCrud for PostgresTrigger {
         DeployedObject::PostgresTrigger { path, parent_path }
     }
 
+    async fn validate_config(
+        &self,
+        db: &DB,
+        config: &Self::TriggerConfigRequest,
+        workspace_id: &str,
+    ) -> Result<()> {
+        ensure_not_under_roles(db, workspace_id, &config.postgres_resource_path).await
+    }
+
+    async fn authorize_set_trigger_mode(
+        &self,
+        _authed: &ApiAuthed,
+        tx: &mut PgConnection,
+        _workspace_id: &str,
+        _path: &str,
+        mode: &TriggerMode,
+    ) -> Result<()> {
+        if *mode != TriggerMode::Disabled {
+            windmill_common::datatable_roles::lock_datatable_streams(tx, false).await?;
+        }
+        Ok(())
+    }
+
     async fn create_trigger(
         &self,
         db: &DB,
@@ -72,6 +96,7 @@ impl TriggerCrud for PostgresTrigger {
         w_id: &str,
         trigger: TriggerData<Self::TriggerConfigRequest>,
     ) -> Result<()> {
+        windmill_common::datatable_roles::lock_datatable_streams(&mut *tx, false).await?;
         let resolved_edited_by = trigger.base.resolve_edited_by(authed);
         let resolved_permissioned_as = trigger.base.resolve_permissioned_as(authed);
         let Self::TriggerConfigRequest {
@@ -161,6 +186,7 @@ impl TriggerCrud for PostgresTrigger {
         path: &str,
         trigger: TriggerData<Self::TriggerConfigRequest>,
     ) -> Result<()> {
+        windmill_common::datatable_roles::lock_datatable_streams(&mut *tx, false).await?;
         let resolved_edited_by = trigger.base.resolve_edited_by(authed);
         let resolved_permissioned_as = trigger.base.resolve_permissioned_as(authed);
         let Self::TriggerConfigRequest {

backend/windmill-trigger-postgres/src/lib.rs

@@ -374,6 +374,32 @@ pub async fn get_raw_postgres_connection(
     Ok(client)
 }
 
+/// A replication stream reads every row of every table whatever the data table's roles grant, so
+/// the two don't mix: a data table under roles takes no triggers or captures, and roles cannot be
+/// turned on while one is enabled on it.
+pub async fn ensure_not_under_roles(
+    db: &DB,
+    w_id: &str,
+    postgres_resource_path: &str,
+) -> Result<()> {
+    let Some(datatable_name) = postgres_resource_path.strip_prefix("datatable://") else {
+        return Ok(());
+    };
+    if windmill_common::workspaces::resolve_governing_datatable(db, w_id, datatable_name)
+        .await?
+        .datatable
+        .permissions
+        .is_some()
+    {
+        return Err(Error::BadRequest(format!(
+            "Data table '{datatable_name}' is under roles, and a Postgres trigger or capture \
+             cannot read one: a replication stream sees every row whatever the roles grant. \
+             Turn its roles off to stream it."
+        )));
+    }
+    Ok(())
+}
+
 pub async fn resolve_postgres_resource(
     authed: &ApiAuthed,
     user_db: Option<UserDB>,
@@ -382,6 +408,7 @@ pub async fn resolve_postgres_resource(
     w_id: &str,
 ) -> Result<Postgres> {
     if let Some(datatable_name) = postgres_resource_path.strip_prefix("datatable://") {
+        ensure_not_under_roles(db, w_id, postgres_resource_path).await?;
         // Trigger connections (publication/slot management + logical replication) run
         // as the dedicated replication user on custom-instance databases.
         let resource_value =

backend/windmill-trigger-postgres/src/listener.rs

@@ -20,7 +20,8 @@ use windmill_common::{
 use windmill_trigger::{listener::ListeningTrigger, trigger_helpers::TriggerJobArgs, Listener};
 
 use super::{
-    drop_publication, get_default_pg_connection, get_raw_postgres_connection,
+    drop_publication, ensure_not_under_roles, get_default_pg_connection,
+    get_raw_postgres_connection,
     handler::drop_logical_replication_slot,
     relation::RelationConverter,
     replication_message::{
@@ -135,8 +136,8 @@ impl PostgresSimpleClient {
 /// Resolves the Postgres resource, validates that the configured publication and
 /// replication slot still exist, and opens a fresh logical replication stream.
 ///
-/// Returns `Error::BadConfig` when the publication or slot is missing (an
-/// unrecoverable misconfiguration). Any other error is treated as transient
+/// Returns `Error::BadConfig` when the publication or slot is missing, or the
+/// data table is under roles (unrecoverable misconfigurations). Any other error is treated as transient
 /// (connection refused, network interruption, ...) and is retried by the caller.
 /// The resource is re-resolved on every call so credential rotations are picked
 /// up across reconnections.
@@ -149,6 +150,14 @@ async fn connect_logical_replication_stream(
     let PostgresConfig { postgres_resource_path, publication_name, replication_slot_name, .. } =
         trigger_config;
 
+    // Retrying cannot lift roles, so this disables the trigger like a missing slot does.
+    ensure_not_under_roles(db, workspace_id, postgres_resource_path)
+        .await
+        .map_err(|e| match e {
+            Error::BadRequest(msg) => Error::BadConfig(msg),
+            e => e,
+        })?;
+
     let database = resolve_postgres_resource(
         authed,
         Some(UserDB::new(db.clone())),

backend/windmill-worker/src/agent_workers.rs

@@ -64,16 +64,25 @@ pub async fn get_ducklake_from_agent_http(
         .await
 }
 
+/// An agent worker authenticates as infrastructure, not as the job's user, so the job id travels
+/// with the request: the server reads the job's owner from it and evaluates the data table's
+/// tenants against them. A worker predating this sends neither, and the server fails it closed on
+/// a data table under roles.
 #[allow(dead_code)]
 pub async fn get_datatable_resource_from_agent_http(
     client: &HttpClient,
     name: &str,
     w_id: &str,
+    role: Option<&str>,
+    job_id: &uuid::Uuid,
 ) -> anyhow::Result<serde_json::Value> {
+    let role_query = role
+        .map(|r| format!("&role={}", urlencoding::encode(r)))
+        .unwrap_or_default();
     client
         .get(&format!(
-            "/api/w/{}/agent_workers/get_datatable_resource/{}",
-            w_id, &name
+            "/api/w/{}/agent_workers/get_datatable_resource/{}?job_id={}{}",
+            w_id, &name, job_id, role_query
         ))
         .await
 }

backend/windmill-worker/src/duckdb_executor.rs

@@ -13,8 +13,8 @@ use windmill_common::error::{to_anyhow, Error, Result};
 use windmill_common::utils::sanitize_string_from_password;
 use windmill_common::worker::{get_memory, to_raw_value, Connection, SqlResultCollectionStrategy};
 use windmill_common::workspaces::{
-    get_datatable_resource_from_db_unchecked, get_ducklake_from_db_unchecked,
-    strip_fork_reserved_attach_args, DucklakeCatalogResourceType,
+    get_datatable_resource_from_db, get_ducklake_from_db_unchecked,
+    strip_fork_reserved_attach_args, DatatableAccess, DucklakeCatalogResourceType,
 };
 use windmill_common::PgDatabase;
 use windmill_object_store::S3_PROXY_LAST_ERRORS_CACHE;
@@ -1494,13 +1494,9 @@ pub async fn do_duckdb(
                 .await?
                 {
                     probe_blocks.extend(q);
-                } else if let Some(q) = transform_attach_datatable(
-                    &query_block,
-                    conn,
-                    &mut hidden_passwords,
-                    &job.workspace_id,
-                )
-                .await?
+                } else if let Some(q) =
+                    transform_attach_datatable(&query_block, conn, &mut hidden_passwords, job)
+                        .await?
                 {
                     probe_blocks.extend(q);
                 } else {
@@ -1575,13 +1571,9 @@ pub async fn do_duckdb(
                 .await?
                 {
                     v.extend(ducklake_query);
-                } else if let Some(datatable_query) = transform_attach_datatable(
-                    &query_block,
-                    conn,
-                    &mut hidden_passwords,
-                    &job.workspace_id,
-                )
-                .await?
+                } else if let Some(datatable_query) =
+                    transform_attach_datatable(&query_block, conn, &mut hidden_passwords, job)
+                        .await?
                 {
                     v.extend(datatable_query);
                 } else {
@@ -2609,33 +2601,79 @@ fn fork_defer_statements(
     Ok(stmts)
 }
 
+struct AttachedDatatable<'a> {
+    /// The data table reference, query string included; a bare `datatable` is `main`.
+    reference: String,
+    alias: &'a str,
+}
+
+/// `ATTACH 'datatable[://<name>][?role=<role>]' AS <alias>`. A bare `datatable` names the default
+/// data table, so the role query string has to be accepted with and without an explicit name. The
+/// reference is split only once the workspace can be read, because a stored name may contain `?`.
+fn parse_attach_datatable(query: &str) -> Option<AttachedDatatable<'_>> {
+    lazy_static::lazy_static! {
+        static ref RE: regex::Regex = regex::Regex::new(
+            r"(?i)ATTACH\s*'datatable(://[^':]+|\?[^':]*)?'\s*AS\s+([^ ;]+)"
+        ).unwrap();
+    }
+    let cap = RE.captures(query)?;
+    let reference = match cap.get(1).map(|m| m.as_str()) {
+        Some(named) if named.starts_with("://") => named[3..].to_string(),
+        Some(query) => format!("main{query}"),
+        None => "main".to_string(),
+    };
+    let alias = cap.get(2).map(|m| m.as_str()).unwrap_or("");
+    Some(AttachedDatatable { reference, alias })
+}
+
 async fn transform_attach_datatable(
     query: &str,
     conn: &Connection,
     hidden_passwords: &mut Arc<Mutex<Vec<String>>>,
-    w_id: &str,
+    job: &MiniPulledJob,
 ) -> Result<Option<Vec<String>>> {
-    lazy_static::lazy_static! {
-        static ref RE: regex::Regex = regex::Regex::new(r"(?i)ATTACH\s*'datatable(://[^':]+)?'\s*AS\s+([^ ;]+)").unwrap();
-    }
-    let Some(cap) = RE.captures(query) else {
+    let Some(attached) = parse_attach_datatable(query) else {
         return Ok(None);
     };
-    let name = cap.get(1).map(|m| &m.as_str()[3..]).unwrap_or("main");
-    let alias_name = cap.get(2).map(|m| m.as_str()).unwrap_or("");
 
+    // A query string that does not parse is refused rather than dropped: attaching under the
+    // default role when the statement asked for another one is the failure this guards.
     let db_resource = match conn {
         Connection::Http(client) => {
-            get_datatable_resource_from_agent_http(client, name, w_id).await?
+            let (name, role) =
+                windmill_common::workspaces::parse_datatable_ref(&attached.reference)?;
+            get_datatable_resource_from_agent_http(client, name, &job.workspace_id, role, &job.id)
+                .await?
+        }
+        Connection::Sql(db) => {
+            let (name, role) = windmill_common::workspaces::parse_datatable_ref_for(
+                db,
+                &job.workspace_id,
+                &attached.reference,
+            )
+            .await?;
+            get_datatable_resource_from_db(
+                db,
+                &job.workspace_id,
+                &name,
+                role.as_deref(),
+                DatatableAccess::PermissionedAs {
+                    permissioned_as: &job.permissioned_as,
+                    email: &job.permissioned_as_email,
+                },
+            )
+            .await?
         }
-        Connection::Sql(db) => get_datatable_resource_from_db_unchecked(db, w_id, name).await?,
     };
 
     if let Some(pwd) = db_resource.get("password").and_then(|p| p.as_str()) {
         hidden_passwords.lock().unwrap().push(pwd.to_string());
     }
 
-    Ok(Some(pg_secret_attach_statements(db_resource, alias_name)?))
+    Ok(Some(pg_secret_attach_statements(
+        db_resource,
+        attached.alias,
+    )?))
 }
 
 // Secret names must be plain identifiers; the hash keeps two aliases distinct even
@@ -2753,6 +2791,45 @@ pub struct Arg {
 mod tests {
     use super::*;
 
+    #[test]
+    fn attach_datatable_parses_name_and_role() {
+        let reference_of = |q: &str| parse_attach_datatable(q).unwrap().reference;
+        let named =
+            parse_attach_datatable("ATTACH 'datatable://sales?role=analytics' AS dt").unwrap();
+        assert_eq!(
+            (named.reference.as_str(), named.alias),
+            ("sales?role=analytics", "dt")
+        );
+        // A bare `datatable` is the default one, and still takes a role.
+        assert_eq!(
+            reference_of("ATTACH 'datatable?role=analytics' AS dt"),
+            "main?role=analytics"
+        );
+        assert_eq!(reference_of("ATTACH 'datatable://sales' AS dt"), "sales");
+        assert_eq!(reference_of("ATTACH 'datatable' AS dt"), "main");
+        assert!(parse_attach_datatable("SELECT 1").is_none());
+        // A stored name can contain `?`, so that is left to the workspace lookup to split.
+        assert_eq!(reference_of("ATTACH 'datatable://a?b' AS dt"), "a?b");
+
+        // The key matches case-insensitively, as the `-- role` annotation does, and a query string
+        // that does not parse is refused rather than attached under the default role.
+        let parse = |q: &str| {
+            windmill_common::workspaces::parse_datatable_ref(&reference_of(q))
+                .map(|(name, role)| (name.to_string(), role.map(str::to_string)))
+        };
+        assert_eq!(
+            parse("ATTACH 'datatable://sales?Role=analytics' AS dt").unwrap(),
+            ("sales".to_string(), Some("analytics".to_string()))
+        );
+        for malformed in [
+            "ATTACH 'datatable://sales?role=' AS dt",
+            "ATTACH 'datatable://sales?role=an;alytics' AS dt",
+            "ATTACH 'datatable://sales?x=1&role=analytics' AS dt",
+        ] {
+            assert!(parse(malformed).is_err(), "silently ignored: {malformed}");
+        }
+    }
+
     #[test]
     fn decode_ffi_error_unescapes_multiline_and_strips_quotes() {
         // Mirror the FFI: JSON-encode the raw DuckDB message, prefix "ERROR ".

backend/windmill-worker/src/pg_executor.rs

@@ -26,9 +26,11 @@ use windmill_common::azure_workload_identity::WORKLOAD_IDENTITY_PASSWORD;
 use windmill_common::error::to_anyhow;
 use windmill_common::error::{self, Error};
 use windmill_common::worker::{
-    to_raw_value, Connection, SqlResultCollectionStrategy, CLOUD_HOSTED,
+    to_raw_value, Connection, SqlAnnotations, SqlResultCollectionStrategy, CLOUD_HOSTED,
+};
+use windmill_common::workspaces::{
+    get_datatable_resource_from_db, parse_datatable_ref, parse_datatable_ref_for, DatatableAccess,
 };
-use windmill_common::workspaces::get_datatable_resource_from_db_unchecked;
 use windmill_common::{PgDatabase, PrepareQueryColumnInfo, PrepareQueryResult, DB};
 use windmill_parser::{Arg, Typ};
 use windmill_parser_sql::{
@@ -680,15 +682,36 @@ pub async fn do_postgresql(
     } else {
         match pg_args.get("database").cloned() {
             Some(Value::String(db_str)) if db_str.starts_with("datatable://") => {
-                let db_str = db_str.trim_start_matches("datatable://");
+                let reference = db_str.trim_start_matches("datatable://");
+                // The annotation wins: a generated query can carry a `?role=` in the reference it
+                // was handed, but only the script's author writes the leading comment block.
+                let annotated = SqlAnnotations::datatable_role(&query)?;
                 Some(match conn {
                     Connection::Http(client) => {
-                        get_datatable_resource_from_agent_http(client, &db_str, &job.workspace_id)
-                            .await?
+                        let (name, uri_role) = parse_datatable_ref(reference)?;
+                        get_datatable_resource_from_agent_http(
+                            client,
+                            name,
+                            &job.workspace_id,
+                            annotated.as_deref().or(uri_role),
+                            &job.id,
+                        )
+                        .await?
                     }
                     Connection::Sql(db) => {
-                        get_datatable_resource_from_db_unchecked(db, &job.workspace_id, &db_str)
-                            .await?
+                        let (name, uri_role) =
+                            parse_datatable_ref_for(db, &job.workspace_id, reference).await?;
+                        get_datatable_resource_from_db(
+                            db,
+                            &job.workspace_id,
+                            &name,
+                            annotated.as_deref().or(uri_role.as_deref()),
+                            DatatableAccess::PermissionedAs {
+                                permissioned_as: &job.permissioned_as,
+                                email: &job.permissioned_as_email,
+                            },
+                        )
+                        .await?
                     }
                 })
             }

cli/src/commands/app/raw_apps.ts

@@ -55,6 +55,8 @@ export interface AppFile {
     tables?: string[];
     datatable?: string;
     schema?: string;
+    /** The role the app uses each data table through, by data table name. */
+    roles?: Record<string, string>;
   };
   // Mirrors granular ACLs on the raw_app path. Synced via /acls/* by
   // applyExtraPermsDiff — never through update_app_raw — so a perm-only