diff --git a/AGENTS.md b/AGENTS.md
index a57215cde5..8c30fed2f9 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -34,8 +34,9 @@ Open-source platform for internal tools, workflows, API integrations, background
   lifecycle lock, managed-object markers, the setup gate, per-cluster roles, fork copy ownership)
   and how to run one locally
 - **Data table viewer roles**: `docs/datatable-viewer-roles.md`, on what a connection naming no
-  role holds: the default role as a floor, where the viewer comes from, why an admin's roles are
-  not the viewer's, the per-role-set logins and the order their rows and logins are written in
+  role holds: the default role as a floor, whose roles are added (the job's own, or its viewer's
+  for an inline app runnable that opts in), why an admin's roles are not added, the per-role-set
+  logins and the order their rows and logins are written in
 - **Enterprise**: `docs/enterprise.md` — EE file conventions and PR workflow
 - **Operator write rights**: `docs/operator-write-rights.md` — which `operator_settings` flags are
   enforced rather than cosmetic, and why a right that is granted-unless-withdrawn needs `Option`
diff --git a/backend/.sqlx/query-711927611ecf0144b6241b8f39f94b79f4ff03bdae60ab746f930f8758fe46bc.json b/backend/.sqlx/query-711927611ecf0144b6241b8f39f94b79f4ff03bdae60ab746f930f8758fe46bc.json
new file mode 100644
index 0000000000..4735c589ac
--- /dev/null
+++ b/backend/.sqlx/query-711927611ecf0144b6241b8f39f94b79f4ff03bdae60ab746f930f8758fe46bc.json
@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE job_perms SET viewer_data_access = true WHERE job_id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "711927611ecf0144b6241b8f39f94b79f4ff03bdae60ab746f930f8758fe46bc"
+}
diff --git a/backend/.sqlx/query-9ec675bd675c9915ed56f6cb91ca848b094b61475409156489319cb9d36ef096.json b/backend/.sqlx/query-9ec675bd675c9915ed56f6cb91ca848b094b61475409156489319cb9d36ef096.json
new file mode 100644
index 0000000000..bd3f00aa7d
--- /dev/null
+++ b/backend/.sqlx/query-9ec675bd675c9915ed56f6cb91ca848b094b61475409156489319cb9d36ef096.json
@@ -0,0 +1,330 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT\n        v2_job_queue.workspace_id,\n        v2_job_queue.id,\n        v2_job.args as \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",\n        v2_job.parent_job,\n        v2_job.created_by,\n        v2_job_queue.started_at,\n        v2_job_queue.runnable_settings_handle,\n        scheduled_for,\n        runnable_path,\n        kind as \"kind: JobKind\",\n        runnable_id as \"runnable_id: ScriptHash\",\n        canceled_reason,\n        canceled_by,\n        permissioned_as,\n        permissioned_as_email,\n        flow_status as \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n        v2_job.tag,\n        script_lang as \"script_lang: ScriptLang\",\n        same_worker,\n        pre_run_error,\n        concurrent_limit,\n        concurrency_time_window_s,\n        flow_innermost_root_job,\n        root_job,\n        timeout,\n        flow_step_id,\n        cache_ttl,\n        cache_ignore_s3_path,\n        v2_job_queue.priority,\n        preprocessed,\n        script_entrypoint_override,\n        trigger,\n        trigger_kind as \"trigger_kind: TriggerKindLabel\",\n        visible_to_owner,\n        NULL as permissioned_as_end_user_email,\n        NULL::boolean as viewer_data_access,\n        job_perms.job_token_scopes\n        FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id\n        LEFT JOIN job_perms ON job_perms.job_id = v2_job_queue.id WHERE v2_job_queue.id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 2,
+        "name": "args: sqlx::types::Json<HashMap<String, Box<RawValue>>>",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 3,
+        "name": "parent_job",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 4,
+        "name": "created_by",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 5,
+        "name": "started_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 6,
+        "name": "runnable_settings_handle",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 7,
+        "name": "scheduled_for",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 8,
+        "name": "runnable_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 9,
+        "name": "kind: JobKind",
+        "type_info": {
+          "Custom": {
+            "name": "job_kind",
+            "kind": {
+              "Enum": [
+                "script",
+                "preview",
+                "flow",
+                "dependencies",
+                "flowpreview",
+                "script_hub",
+                "identity",
+                "flowdependencies",
+                "http",
+                "graphql",
+                "postgresql",
+                "noop",
+                "appdependencies",
+                "deploymentcallback",
+                "singlestepflow",
+                "flowscript",
+                "flownode",
+                "appscript",
+                "aiagent",
+                "unassigned_script",
+                "unassigned_flow",
+                "unassigned_singlestepflow"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 10,
+        "name": "runnable_id: ScriptHash",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 11,
+        "name": "canceled_reason",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 12,
+        "name": "canceled_by",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 13,
+        "name": "permissioned_as",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 14,
+        "name": "permissioned_as_email",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 15,
+        "name": "flow_status: sqlx::types::Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 16,
+        "name": "tag",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 17,
+        "name": "script_lang: ScriptLang",
+        "type_info": {
+          "Custom": {
+            "name": "script_lang",
+            "kind": {
+              "Enum": [
+                "python3",
+                "deno",
+                "go",
+                "bash",
+                "postgresql",
+                "nativets",
+                "bun",
+                "mysql",
+                "bigquery",
+                "snowflake",
+                "graphql",
+                "powershell",
+                "mssql",
+                "php",
+                "bunnative",
+                "rust",
+                "ansible",
+                "csharp",
+                "oracledb",
+                "nu",
+                "java",
+                "duckdb",
+                "ruby",
+                "rlang",
+                "dbt"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 18,
+        "name": "same_worker",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 19,
+        "name": "pre_run_error",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 20,
+        "name": "concurrent_limit",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 21,
+        "name": "concurrency_time_window_s",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 22,
+        "name": "flow_innermost_root_job",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 23,
+        "name": "root_job",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 24,
+        "name": "timeout",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 25,
+        "name": "flow_step_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 26,
+        "name": "cache_ttl",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 27,
+        "name": "cache_ignore_s3_path",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 28,
+        "name": "priority",
+        "type_info": "Int2"
+      },
+      {
+        "ordinal": 29,
+        "name": "preprocessed",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 30,
+        "name": "script_entrypoint_override",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 31,
+        "name": "trigger",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 32,
+        "name": "trigger_kind: TriggerKindLabel",
+        "type_info": {
+          "Custom": {
+            "name": "job_trigger_kind",
+            "kind": {
+              "Enum": [
+                "webhook",
+                "http",
+                "websocket",
+                "kafka",
+                "email",
+                "nats",
+                "schedule",
+                "app",
+                "ui",
+                "postgres",
+                "sqs",
+                "gcp",
+                "mqtt",
+                "nextcloud",
+                "google",
+                "ci_test",
+                "github",
+                "azure",
+                "asset",
+                "freshness",
+                "amqp"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 33,
+        "name": "visible_to_owner",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 34,
+        "name": "permissioned_as_end_user_email",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 35,
+        "name": "viewer_data_access",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 36,
+        "name": "job_token_scopes",
+        "type_info": "TextArray"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      true,
+      true,
+      false,
+      true,
+      true,
+      false,
+      true,
+      false,
+      true,
+      true,
+      true,
+      false,
+      false,
+      true,
+      false,
+      true,
+      false,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      false,
+      null,
+      null,
+      true
+    ]
+  },
+  "hash": "9ec675bd675c9915ed56f6cb91ca848b094b61475409156489319cb9d36ef096"
+}
diff --git a/backend/.sqlx/query-fa62a55dd0a635995390ad4189b847b91388b6699f16afb57d8147d7e9712220.json b/backend/.sqlx/query-fa62a55dd0a635995390ad4189b847b91388b6699f16afb57d8147d7e9712220.json
new file mode 100644
index 0000000000..bfda06948d
--- /dev/null
+++ b/backend/.sqlx/query-fa62a55dd0a635995390ad4189b847b91388b6699f16afb57d8147d7e9712220.json
@@ -0,0 +1,337 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE v2_job_status SET\n            flow_status = JSONB_SET(\n                JSONB_SET(v2_job_status.flow_status, $2::TEXT[], $3),\n                $4::TEXT[], $5\n            ) - $6::TEXT[],\n            flow_leaf_jobs = CASE\n                WHEN $7::TEXT IS NULL\n                    OR COALESCE(v2_job.flow_innermost_root_job, v2_job_status.id) <> v2_job_status.id\n                THEN v2_job_status.flow_leaf_jobs\n                ELSE JSONB_SET(COALESCE(v2_job_status.flow_leaf_jobs, '{}'::JSONB), ARRAY[$7::TEXT], $8) END\n        FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id\n            LEFT JOIN job_perms ON job_perms.job_id = v2_job_queue.id\n        WHERE v2_job_status.id = $1 AND v2_job_queue.id = $1\n        RETURNING\n            v2_job_queue.workspace_id,\n            v2_job_queue.id,\n            v2_job.args as \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",\n            v2_job.parent_job,\n            v2_job.created_by,\n            v2_job_queue.started_at,\n            v2_job_queue.runnable_settings_handle,\n            v2_job_queue.scheduled_for,\n            v2_job.runnable_path,\n            v2_job.kind as \"kind: JobKind\",\n            v2_job.runnable_id as \"runnable_id: ScriptHash\",\n            v2_job_queue.canceled_reason,\n            v2_job_queue.canceled_by,\n            v2_job.permissioned_as,\n            v2_job.permissioned_as_email,\n            v2_job_status.flow_status as \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n            v2_job.tag,\n            v2_job.script_lang as \"script_lang: ScriptLang\",\n            v2_job.same_worker,\n            v2_job.pre_run_error,\n            v2_job.concurrent_limit,\n            v2_job.concurrency_time_window_s,\n            v2_job.flow_innermost_root_job,\n            v2_job.root_job,\n            v2_job.timeout,\n            v2_job.flow_step_id,\n            v2_job.cache_ttl,\n            v2_job_queue.cache_ignore_s3_path,\n            v2_job_queue.priority,\n            v2_job.preprocessed,\n            v2_job.script_entrypoint_override,\n            v2_job.trigger,\n            v2_job.trigger_kind as \"trigger_kind: TriggerKindLabel\",\n            v2_job.visible_to_owner,\n            NULL as permissioned_as_end_user_email,\n            NULL::boolean as viewer_data_access,\n            job_perms.job_token_scopes",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 2,
+        "name": "args: sqlx::types::Json<HashMap<String, Box<RawValue>>>",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 3,
+        "name": "parent_job",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 4,
+        "name": "created_by",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 5,
+        "name": "started_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 6,
+        "name": "runnable_settings_handle",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 7,
+        "name": "scheduled_for",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 8,
+        "name": "runnable_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 9,
+        "name": "kind: JobKind",
+        "type_info": {
+          "Custom": {
+            "name": "job_kind",
+            "kind": {
+              "Enum": [
+                "script",
+                "preview",
+                "flow",
+                "dependencies",
+                "flowpreview",
+                "script_hub",
+                "identity",
+                "flowdependencies",
+                "http",
+                "graphql",
+                "postgresql",
+                "noop",
+                "appdependencies",
+                "deploymentcallback",
+                "singlestepflow",
+                "flowscript",
+                "flownode",
+                "appscript",
+                "aiagent",
+                "unassigned_script",
+                "unassigned_flow",
+                "unassigned_singlestepflow"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 10,
+        "name": "runnable_id: ScriptHash",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 11,
+        "name": "canceled_reason",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 12,
+        "name": "canceled_by",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 13,
+        "name": "permissioned_as",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 14,
+        "name": "permissioned_as_email",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 15,
+        "name": "flow_status: sqlx::types::Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 16,
+        "name": "tag",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 17,
+        "name": "script_lang: ScriptLang",
+        "type_info": {
+          "Custom": {
+            "name": "script_lang",
+            "kind": {
+              "Enum": [
+                "python3",
+                "deno",
+                "go",
+                "bash",
+                "postgresql",
+                "nativets",
+                "bun",
+                "mysql",
+                "bigquery",
+                "snowflake",
+                "graphql",
+                "powershell",
+                "mssql",
+                "php",
+                "bunnative",
+                "rust",
+                "ansible",
+                "csharp",
+                "oracledb",
+                "nu",
+                "java",
+                "duckdb",
+                "ruby",
+                "rlang",
+                "dbt"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 18,
+        "name": "same_worker",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 19,
+        "name": "pre_run_error",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 20,
+        "name": "concurrent_limit",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 21,
+        "name": "concurrency_time_window_s",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 22,
+        "name": "flow_innermost_root_job",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 23,
+        "name": "root_job",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 24,
+        "name": "timeout",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 25,
+        "name": "flow_step_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 26,
+        "name": "cache_ttl",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 27,
+        "name": "cache_ignore_s3_path",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 28,
+        "name": "priority",
+        "type_info": "Int2"
+      },
+      {
+        "ordinal": 29,
+        "name": "preprocessed",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 30,
+        "name": "script_entrypoint_override",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 31,
+        "name": "trigger",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 32,
+        "name": "trigger_kind: TriggerKindLabel",
+        "type_info": {
+          "Custom": {
+            "name": "job_trigger_kind",
+            "kind": {
+              "Enum": [
+                "webhook",
+                "http",
+                "websocket",
+                "kafka",
+                "email",
+                "nats",
+                "schedule",
+                "app",
+                "ui",
+                "postgres",
+                "sqs",
+                "gcp",
+                "mqtt",
+                "nextcloud",
+                "google",
+                "ci_test",
+                "github",
+                "azure",
+                "asset",
+                "freshness",
+                "amqp"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 33,
+        "name": "visible_to_owner",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 34,
+        "name": "permissioned_as_end_user_email",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 35,
+        "name": "viewer_data_access",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 36,
+        "name": "job_token_scopes",
+        "type_info": "TextArray"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "TextArray",
+        "Jsonb",
+        "TextArray",
+        "Jsonb",
+        "TextArray",
+        "Text",
+        "Jsonb"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      true,
+      true,
+      false,
+      true,
+      true,
+      false,
+      true,
+      false,
+      true,
+      true,
+      true,
+      false,
+      false,
+      true,
+      false,
+      true,
+      false,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      false,
+      null,
+      null,
+      true
+    ]
+  },
+  "hash": "fa62a55dd0a635995390ad4189b847b91388b6699f16afb57d8147d7e9712220"
+}
diff --git a/backend/.sqlx/query-fd04f65015069792f23edbfce320af6e69ee32b27a8f0a020e251eb2c79e43cd.json b/backend/.sqlx/query-fd04f65015069792f23edbfce320af6e69ee32b27a8f0a020e251eb2c79e43cd.json
new file mode 100644
index 0000000000..4801a9b5da
--- /dev/null
+++ b/backend/.sqlx/query-fd04f65015069792f23edbfce320af6e69ee32b27a8f0a020e251eb2c79e43cd.json
@@ -0,0 +1,71 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT email, username, is_admin, is_operator, groups, folders, end_user_email,\n            job_token_scopes, viewer_data_access\n        FROM job_perms WHERE job_id = $1 AND workspace_id = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "email",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "username",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "is_admin",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 3,
+        "name": "is_operator",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 4,
+        "name": "groups",
+        "type_info": "TextArray"
+      },
+      {
+        "ordinal": 5,
+        "name": "folders",
+        "type_info": "JsonbArray"
+      },
+      {
+        "ordinal": 6,
+        "name": "end_user_email",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 7,
+        "name": "job_token_scopes",
+        "type_info": "TextArray"
+      },
+      {
+        "ordinal": 8,
+        "name": "viewer_data_access",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      true,
+      true,
+      false
+    ]
+  },
+  "hash": "fd04f65015069792f23edbfce320af6e69ee32b27a8f0a020e251eb2c79e43cd"
+}
diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt
index bbd2895dd5..42fd37a116 100644
--- a/backend/ee-repo-ref.txt
+++ b/backend/ee-repo-ref.txt
@@ -1 +1 @@
-196abc042c691ea16280ee10b89fb3b2fd527236
+0ffb633473c5465513576afc6f54d18f93c7f1e1
diff --git a/backend/migrations/20261013165527_job_perms_viewer_data_access.down.sql b/backend/migrations/20261013165527_job_perms_viewer_data_access.down.sql
new file mode 100644
index 0000000000..6741203ade
--- /dev/null
+++ b/backend/migrations/20261013165527_job_perms_viewer_data_access.down.sql
@@ -0,0 +1 @@
+ALTER TABLE job_perms DROP COLUMN IF EXISTS viewer_data_access;
diff --git a/backend/migrations/20261013165527_job_perms_viewer_data_access.up.sql b/backend/migrations/20261013165527_job_perms_viewer_data_access.up.sql
new file mode 100644
index 0000000000..5d6a41ece4
--- /dev/null
+++ b/backend/migrations/20261013165527_job_perms_viewer_data_access.up.sql
@@ -0,0 +1,4 @@
+-- Set by the server on a job pushed for an app runnable whose policy says it reads data as its
+-- viewer: data table connections that name no role and object storage permission rules then go by
+-- end_user_email instead of the job's own identity.
+ALTER TABLE job_perms ADD COLUMN IF NOT EXISTS viewer_data_access BOOLEAN NOT NULL DEFAULT false;
diff --git a/backend/summarized_schema.txt b/backend/summarized_schema.txt
index ac07dc06d6..fa1c049746 100644
--- a/backend/summarized_schema.txt
+++ b/backend/summarized_schema.txt
@@ -132,7 +132,7 @@ input: id(uuid), workspace_id(char), runnable_id(char), runnable_type(runnable_t
   FK: (workspace_id) -> workspace(id)
 instance_group: name(char), summary(char), id(char), scim_display_name(char), external_id(char)
 job_logs: job_id(uuid), workspace_id(char), created_at(ts), logs(text), log_offset(int), log_file_index(text[])
-job_perms: job_id(uuid), email(char), username(char), is_admin(bool), is_operator(bool), created_at(ts), workspace_id(char), groups(text[]), folders(jsonb[]), end_user_email(char), job_token_scopes(text[])
+job_perms: job_id(uuid), email(char), username(char), is_admin(bool), is_operator(bool), created_at(ts), workspace_id(char), groups(text[]), folders(jsonb[]), end_user_email(char), job_token_scopes(text[]), viewer_data_access(bool)
 job_resolution: job_id(uuid), workspace_id(char), resolved_at(ts), resolved_by(char), note(text), automatic(bool)
 job_result_stream: job_id(uuid), workspace_id(text), stream(text)
 job_result_stream_v2: job_id(uuid), workspace_id(text), stream(text), idx(int)
diff --git a/backend/tests/end_user_email.rs b/backend/tests/end_user_email.rs
index 0d8ff7be9a..15eb9a64f1 100644
--- a/backend/tests/end_user_email.rs
+++ b/backend/tests/end_user_email.rs
@@ -509,3 +509,89 @@ async fn test_raw_app_wm_end_user_email(db: Pool<Postgres>) -> anyhow::Result<()
 
     Ok(())
 }
+
+/// Connecting to data tables as the viewer is honored on inline runnables only: a path runnable
+/// is someone else's script, which must not read with the viewer's roles because the app's
+/// policy entry for it says so.
+#[sqlx::test(fixtures("base", "end_user_email"))]
+async fn test_app_viewer_data_access_is_stamped_on_inline_runnables_only(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    set_jwt_secret().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    let app_path = "f/test/viewer_data_access_app";
+    let script_path = "script/f/test/get_end_user_email";
+
+    let resp = authed(
+        client().post(format!(
+            "http://localhost:{port}/api/w/test-workspace/apps/create"
+        )),
+        SAME_WS_TOKEN,
+    )
+    .json(&json!({
+        "path": app_path,
+        "summary": "",
+        "value": { "type": "app", "grid": [], "subgrids": {}, "hiddenInlineScripts": [] },
+        "policy": {
+            "execution_mode": "anonymous",
+            "on_behalf_of": null,
+            "on_behalf_of_email": null,
+            "triggerables_v2": {
+                "rawscript/894d2f85ea2411e777e7d5787366c9fec4c1a7a5f83482748f1d129196c9f54f": {
+                    "static_inputs": {}, "one_of_inputs": {}, "viewer_data_access": true
+                },
+                (script_path): {
+                    "static_inputs": {}, "one_of_inputs": {}, "viewer_data_access": true
+                }
+            }
+        }
+    }))
+    .send()
+    .await?;
+    anyhow::ensure!(resp.status().is_success(), "create app: {}", resp.text().await?);
+
+    let execute = |payload: serde_json::Value| async move {
+        let resp = authed(
+            client().post(format!(
+                "http://localhost:{port}/api/w/test-workspace/apps_u/execute_component/{app_path}"
+            )),
+            SAME_WS_TOKEN,
+        )
+        .json(&payload)
+        .send()
+        .await?;
+        anyhow::ensure!(resp.status().is_success(), "execute: {}", resp.text().await?);
+        Ok::<_, anyhow::Error>(resp.text().await?.trim_matches('"').parse::<uuid::Uuid>()?)
+    };
+    let as_viewer = |job_id: uuid::Uuid| {
+        let db = db.clone();
+        async move {
+            sqlx::query_scalar::<_, bool>(
+                "SELECT viewer_data_access FROM job_perms WHERE job_id = $1",
+            )
+            .bind(job_id)
+            .fetch_one(&db)
+            .await
+        }
+    };
+
+    let inline = execute(json!({
+        "args": {},
+        "component": "get_email",
+        "raw_code": {
+            "language": "bun",
+            "content": "export function main() { return process.env.WM_END_USER_EMAIL || \"\"; }"
+        }
+    }))
+    .await?;
+    assert!(as_viewer(inline).await?);
+
+    let path = execute(json!({ "args": {}, "component": "run_script", "path": script_path })).await?;
+    assert!(
+        !as_viewer(path).await?,
+        "a path runnable connected as the viewer because its policy entry said so"
+    );
+    Ok(())
+}
diff --git a/backend/tests/job_s3_viewer_data_access.rs b/backend/tests/job_s3_viewer_data_access.rs
new file mode 100644
index 0000000000..a68b695259
--- /dev/null
+++ b/backend/tests/job_s3_viewer_data_access.rs
@@ -0,0 +1,112 @@
+//! A job whose `job_perms.viewer_data_access` is set reads workspace object storage as its viewer:
+//! the advanced permission rules are checked against the viewer, not the job's own identity, and
+//! such a job with no viewer is refused rather than checked as itself.
+//!
+//! Advanced S3 permissions are an enterprise feature.
+#![cfg(all(feature = "enterprise", feature = "private", feature = "parquet"))]
+
+use serde_json::json;
+use sqlx::{Pool, Postgres};
+use uuid::Uuid;
+use windmill_test_utils::*;
+
+/// A job of `test-user-3`, run for `viewer`, reading data as that viewer or not.
+async fn job_token(
+    db: &Pool<Postgres>,
+    viewer: Option<&str>,
+    viewer_data_access: bool,
+) -> anyhow::Result<String> {
+    let id = Uuid::new_v4();
+    sqlx::query(
+        "INSERT INTO v2_job (id, workspace_id, created_by, permissioned_as, permissioned_as_email,
+            kind, script_lang, runnable_path, tag)
+        VALUES ($1, 'test-workspace', 'test-user-3', 'u/test-user-3', 'test3@windmill.dev',
+            'script', 'deno', 'u/test-user-3/app_runnable', 'deno')",
+    )
+    .bind(id)
+    .execute(db)
+    .await?;
+    sqlx::query(
+        "INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups,
+            workspace_id, end_user_email, viewer_data_access)
+        VALUES ($1, 'test3@windmill.dev', 'test-user-3', false, false, '{}', '{}',
+            'test-workspace', $2, $3)",
+    )
+    .bind(id)
+    .bind(viewer)
+    .bind(viewer_data_access)
+    .execute(db)
+    .await?;
+    Ok(windmill_common::auth::create_token_for_owner(
+        db,
+        "test-workspace",
+        "u/test-user-3",
+        "ephemeral-script",
+        300,
+        "test3@windmill.dev",
+        &id,
+        None,
+        None,
+    )
+    .await?)
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn a_job_reading_data_as_its_viewer_gets_the_viewers_storage_paths(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    set_jwt_secret().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+
+    let storage_dir = tempfile::tempdir()?;
+    for user in ["test-user-2", "test-user-3"] {
+        let dir = storage_dir.path().join("u").join(user);
+        std::fs::create_dir_all(&dir)?;
+        std::fs::write(dir.join("file.txt"), user)?;
+    }
+    sqlx::query("UPDATE workspace_settings SET large_file_storage = $1 WHERE workspace_id = $2")
+        .bind(json!({
+            "type": "FilesystemStorage",
+            "root_path": storage_dir.path().to_string_lossy(),
+            "public_resource": null,
+            "advanced_permissions": [{ "pattern": "u/{username}/**", "allow": "read" }]
+        }))
+        .bind("test-workspace")
+        .execute(&db)
+        .await?;
+
+    let read = |token: String, user: &'static str| async move {
+        let resp = reqwest::Client::new()
+            .get(format!(
+                "http://localhost:{port}/api/w/test-workspace/job_helpers/download_s3_file"
+            ))
+            .query(&[("file_key", format!("u/{user}/file.txt"))])
+            .bearer_auth(token)
+            .send()
+            .await?;
+        Ok::<_, anyhow::Error>(resp.status().is_success())
+    };
+
+    let own = job_token(&db, Some("test2@windmill.dev"), false).await?;
+    assert!(read(own.clone(), "test-user-3").await?);
+    assert!(
+        !read(own, "test-user-2").await?,
+        "a job not reading as its viewer reached the viewer's paths"
+    );
+
+    let as_viewer = job_token(&db, Some("test2@windmill.dev"), true).await?;
+    assert!(read(as_viewer.clone(), "test-user-2").await?);
+    assert!(
+        !read(as_viewer, "test-user-3").await?,
+        "a job reading as its viewer kept its own identity's paths"
+    );
+
+    let no_viewer = job_token(&db, None, true).await?;
+    assert!(
+        !read(no_viewer, "test-user-3").await?,
+        "a job reading as a viewer it does not have was checked as itself"
+    );
+    Ok(())
+}
diff --git a/backend/windmill-api-integration-tests/tests/datatable_roles.rs b/backend/windmill-api-integration-tests/tests/datatable_roles.rs
index 023d9f20ed..0acba27c41 100644
--- a/backend/windmill-api-integration-tests/tests/datatable_roles.rs
+++ b/backend/windmill-api-integration-tests/tests/datatable_roles.rs
@@ -1850,10 +1850,12 @@ async fn cleanup_waits_out_a_save_racing_it_for_an_instance_database(
 
 #[cfg(all(feature = "private", feature = "enterprise"))]
 #[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
-async fn a_connection_naming_no_role_holds_the_default_one_and_the_viewers(
+async fn a_connection_naming_no_role_holds_the_default_one_and_its_identitys(
     db: Pool<Postgres>,
 ) -> anyhow::Result<()> {
-    use windmill_common::workspaces::{get_datatable_resource_from_db, DatatableAccess};
+    use windmill_common::workspaces::{
+        get_datatable_resource_from_db, DatatableAccess, DatatableConnectAs,
+    };
 
     initialize_tracing().await;
     let server = ApiServer::start(db.clone()).await?;
@@ -1869,20 +1871,25 @@ async fn a_connection_naming_no_role_holds_the_default_one_and_the_viewers(
          JOIN pg_roles r ON r.oid = am.roleid JOIN pg_roles m ON m.oid = am.member
          WHERE r.rolname = $1 AND m.rolname LIKE 'wm\\_set\\_%'";
 
-    // The job is the app publisher's in every case: only the viewer the server recorded differs.
-    let connect = |role: Option<&'static str>, viewer: Option<&'static str>| {
+    let connect_job = |owner: &'static str, role: Option<&'static str>, connect_as| {
+        let (permissioned_as, email) = match owner {
+            "test-user" => ("u/test-user", "test@windmill.dev"),
+            "test-user-2" => ("u/test-user-2", "test2@windmill.dev"),
+            _ => ("u/test-user-3", "test3@windmill.dev"),
+        };
         get_datatable_resource_from_db(
             &db,
             "test-workspace",
             "main",
             role,
-            DatatableAccess::PermissionedAs {
-                permissioned_as: "u/test-user",
-                email: "test@windmill.dev",
-                end_user_email: viewer,
-            },
+            DatatableAccess::PermissionedAs { permissioned_as, email, connect_as },
         )
     };
+    // An app published by the workspace admin `test-user`, run by `viewer`, whose runnable
+    // connects as its viewer.
+    let connect = |role: Option<&'static str>, viewer: Option<&'static str>| {
+        connect_job("test-user", role, DatatableConnectAs::Viewer(viewer))
+    };
     let set_permissions = |default_role: String, roles: Value| {
         sqlx::query(
             "UPDATE workspace_settings
@@ -1953,11 +1960,25 @@ async fn a_connection_naming_no_role_holds_the_default_one_and_the_viewers(
             expected
         );
 
-        // An admin may use every role, the one owning the tables included. As a viewer they add
-        // what names them, which here is nothing; nor does a job with no viewer recorded.
-        for viewer in [Some("test@windmill.dev"), None] {
-            assert_eq!(connect(None, viewer).await?["user"], names[0].as_str());
-        }
+        // A runnable that does not connect as its viewer holds its own identity's roles, whoever
+        // views it: here the publisher, an admin, who may use every role but adds only what
+        // names them, which is nothing.
+        assert_eq!(
+            connect_job("test-user", None, DatatableConnectAs::Job).await?["user"],
+            names[0].as_str(),
+            "the viewer's roles reached a job that does not connect as its viewer"
+        );
+        // A job with no viewer, a schedule say, holds its owner's roles.
+        assert_eq!(
+            connect_job("test-user-2", None, DatatableConnectAs::Job).await?,
+            all
+        );
+        assert_eq!(
+            connect(None, Some("test@windmill.dev")).await?["user"],
+            names[0].as_str()
+        );
+        // Connecting as a viewer it does not have is refused rather than made as the job.
+        assert!(connect(None, None).await.is_err());
         // A named role is that role alone, whoever is viewing.
         let named: &'static str = Box::leak(names[1].clone().into_boxed_str());
         assert_eq!(
@@ -1978,29 +1999,28 @@ async fn a_connection_naming_no_role_holds_the_default_one_and_the_viewers(
             two,
             "no default changes nothing for a viewer the roles name"
         );
+        assert_eq!(
+            connect_job("test-user-3", None, DatatableConnectAs::Job).await?,
+            two
+        );
+        assert_eq!(
+            connect_job("test-user", None, DatatableConnectAs::Job).await?["user"],
+            names[0].as_str(),
+            "the role open to everyone names the admin too"
+        );
         assert!(connect(None, None).await.is_err());
         // The same holds for a default the job's own identity is not a tenant of.
         let mut closed = roles.clone();
         closed[ids[0].as_str()] = json!({"tenants": []});
         set_permissions(ids[0].clone(), closed).await?;
-        let as_member = |viewer: Option<&'static str>| {
-            get_datatable_resource_from_db(
-                &db,
-                "test-workspace",
-                "main",
-                None,
-                DatatableAccess::PermissionedAs {
-                    permissioned_as: "u/test-user-3",
-                    email: "test3@windmill.dev",
-                    end_user_email: viewer,
-                },
-            )
-        };
         assert_eq!(
-            as_member(Some("test3@windmill.dev")).await?["user"],
+            connect_job("test-user-3", None, DatatableConnectAs::Job).await?["user"],
+            names[1].as_str()
+        );
+        assert_eq!(
+            connect(None, Some("test3@windmill.dev")).await?["user"],
             names[1].as_str()
         );
-        assert!(as_member(None).await.is_err());
         set_permissions(ids[0].clone(), roles).await?;
 
         // A set login nobody connects with any more is dropped, a recent one is kept, and a
diff --git a/backend/windmill-api-jobs/src/execution.rs b/backend/windmill-api-jobs/src/execution.rs
index c6c2a1fede..330e96c2fb 100644
--- a/backend/windmill-api-jobs/src/execution.rs
+++ b/backend/windmill-api-jobs/src/execution.rs
@@ -266,6 +266,22 @@ pub async fn run_end_user_email(
     }
 }
 
+/// [`run_end_user_email`] of an inline run of the caller's own code, and whether its data table
+/// connections are made as that end user: they are when the calling job's are, since the code is
+/// that job's own.
+pub async fn run_inline_preview_end_user(
+    db: &DB,
+    w_id: &str,
+    authed: &ApiAuthed,
+) -> error::Result<(Option<String>, bool)> {
+    match authed.job_id {
+        Some(job_id) => Ok(windmill_common::auth::get_job_perms(db, &job_id, w_id)
+            .await?
+            .map_or((None, false), |p| (p.end_user_email, p.viewer_data_access))),
+        None => Ok((Some(authed.email.clone()), false)),
+    }
+}
+
 #[cfg(feature = "enterprise")]
 pub async fn check_license_key_valid() -> error::Result<()> {
     use windmill_common::ee_oss::LICENSE_KEY_VALID;
diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs
index db8a38f2dd..b3dcc418cd 100644
--- a/backend/windmill-api/src/apps.rs
+++ b/backend/windmill-api/src/apps.rs
@@ -481,6 +481,12 @@ pub struct PolicyTriggerableInputs {
     sensitive_inputs: Vec<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     tag: Option<String>,
+    /// The runnable reads data as the app's viewer rather than as the app's identity: data table
+    /// connections that name no role, and object storage advanced permission rules. Honored for
+    /// inline runnables only: their code is deployed with the app, so it is the publisher who
+    /// chose to read with the viewer's rights.
+    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
+    viewer_data_access: bool,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
@@ -4501,6 +4507,7 @@ async fn execute_component(
     // on the inline script is honored. In any other case we must read the
     // tag from the deployed policy and ignore the request body.
     let is_preview = payload.force_viewer_static_fields.is_some();
+    let is_inline = payload.raw_code.is_some();
 
     // Preview mode runs request-supplied code as a `Viewer`-mode job (the
     // app-editor equivalent of `/jobs/run/preview`), so it enforces the same
@@ -4595,6 +4602,8 @@ async fn execute_component(
                 delete_after_secs: force_viewer_delete_after_secs,
                 sensitive_inputs: force_viewer_sensitive_inputs.unwrap_or_default(),
                 tag: None,
+                // A preview runs as the caller, whose roles these are either way.
+                viewer_data_access: false,
             },
             None,
         ),
@@ -4961,6 +4970,15 @@ async fn execute_component(
     )
     .await?;
 
+    if policy_triggerables.viewer_data_access && is_inline {
+        sqlx::query!(
+            "UPDATE job_perms SET viewer_data_access = true WHERE job_id = $1",
+            uuid
+        )
+        .execute(&mut *tx)
+        .await?;
+    }
+
     // Apply runnable query parameters if provided
     if let Some(ref run_query) = payload.run_query_params {
         if let Some(flow_path) = flow_path.as_deref() {
@@ -6042,11 +6060,10 @@ async fn app_load_table_count(
     let file_query = app_s3_file_query(file_key.clone(), inner.storage.clone(), sig);
     let job_authed =
         app_s3_on_behalf_and_provenance(&db, &path, &w_id, &opt_authed, &file_query).await?;
-    let resp =
-        crate::job_helpers_oss::load_table_count_internal(
-            job_authed, &db, None, &w_id, file_key, inner,
-        )
-            .await?;
+    let resp = crate::job_helpers_oss::load_table_count_internal(
+        job_authed, &db, None, &w_id, file_key, inner,
+    )
+    .await?;
     Ok(Json(resp).into_response())
 }
 
diff --git a/backend/windmill-api/src/apps_raw_policy.gen.js b/backend/windmill-api/src/apps_raw_policy.gen.js
index 1d7c5a48fc..282cd48359 100644
--- a/backend/windmill-api/src/apps_raw_policy.gen.js
+++ b/backend/windmill-api/src/apps_raw_policy.gen.js
@@ -4,4 +4,4 @@
 //
 // Prepended to the raw-app bundle job (see apps_raw_bundle.rs), which calls
 // `__wmillAppPolicy.updateRawAppPolicy` from its own module scope.
-var __wmillAppPolicy=(()=>{var g=Object.create;var i=Object.defineProperty;var b=Object.getOwnPropertyDescriptor;var m=Object.getOwnPropertyNames;var A=Object.getPrototypeOf,S=Object.prototype.hasOwnProperty;var x=(e=>typeof require<"u"?require:typeof Proxy<"u"?new Proxy(e,{get:(t,p)=>(typeof require<"u"?require:t)[p]}):e)(function(e){if(typeof require<"u")return require.apply(this,arguments);throw Error('Dynamic require of "'+e+'" is not supported')});var f=(e,t)=>{for(var p in t)i(e,p,{get:t[p],enumerable:!0})},s=(e,t,p,a)=>{if(t&&typeof t=="object"||typeof t=="function")for(let r of m(t))!S.call(e,r)&&r!==p&&i(e,r,{get:()=>t[r],enumerable:!(a=b(t,r))||a.enumerable});return e};var h=(e,t,p)=>(p=e!=null?g(A(e)):{},s(t||!e||!e.__esModule?i(p,"default",{value:e,enumerable:!0}):p,e)),R=e=>s(i({},"__esModule",{value:!0}),e);var v={};f(v,{updateRawAppPolicy:()=>d});function u(e){return Object.fromEntries(Object.entries(e??{}).filter(([t,p])=>p.type=="static").map(([t,p])=>[t,p.value]))}async function c(e){try{let t=new TextEncoder().encode(e),p=await crypto.subtle.digest("SHA-256",t);return Array.from(new Uint8Array(p)).map(n=>n.toString(16).padStart(2,"0")).join("")}catch{let{Sha256:t}=await import("@aws-crypto/sha256-js"),p=new t;return p.update(e??""),Array.from(await p.digest()).map(n=>n.toString(16).padStart(2,"0")).join("")}}function l(e){return e?.type=="runnableByPath"||e?.type=="path"}function y(e){return e?.type=="runnableByName"||e?.type=="inline"}async function d(e,t){let p=(await Promise.all(Object.entries(e).map(async([n,o])=>await _(n,o,o?.fields??{})))).filter(n=>n!=null),a=Object.fromEntries(p);return{...t,triggerables_v2:a}}function I(e,t){let p={};typeof e.delete_after_secs=="number"&&e.delete_after_secs>=0&&(p.delete_after_secs=e.delete_after_secs);let a=Object.entries(t).map(([r,n])=>n.sensitive?r:void 0).filter(Boolean);return a.length>0&&(p.sensitive_inputs=a),e.inlineScript?.tag&&(p.tag=e.inlineScript.tag),p}async function _(e,t,p){let a=u(p),r=Object.entries(p).map(([n,o])=>o.allowUserResources?n:void 0).filter(Boolean);if(y(t)){let n=await c(t.inlineScript?.content);return[`${e}:rawscript/${n}`,{static_inputs:a,one_of_inputs:{},allow_user_resources:r,...I(t,p)}]}else if(l(t)){let n=t.runType!=="hubscript"?t.runType:"script";return[`${e}:${n}/${t.path}`,{static_inputs:a,one_of_inputs:{},allow_user_resources:r,...I(t,p)}]}}return R(v);})();
+var __wmillAppPolicy=(()=>{var g=Object.create;var i=Object.defineProperty;var b=Object.getOwnPropertyDescriptor;var m=Object.getOwnPropertyNames;var A=Object.getPrototypeOf,S=Object.prototype.hasOwnProperty;var x=(e=>typeof require<"u"?require:typeof Proxy<"u"?new Proxy(e,{get:(t,p)=>(typeof require<"u"?require:t)[p]}):e)(function(e){if(typeof require<"u")return require.apply(this,arguments);throw Error('Dynamic require of "'+e+'" is not supported')});var f=(e,t)=>{for(var p in t)i(e,p,{get:t[p],enumerable:!0})},c=(e,t,p,a)=>{if(t&&typeof t=="object"||typeof t=="function")for(let r of m(t))!S.call(e,r)&&r!==p&&i(e,r,{get:()=>t[r],enumerable:!(a=b(t,r))||a.enumerable});return e};var h=(e,t,p)=>(p=e!=null?g(A(e)):{},c(t||!e||!e.__esModule?i(p,"default",{value:e,enumerable:!0}):p,e)),R=e=>c(i({},"__esModule",{value:!0}),e);var v={};f(v,{updateRawAppPolicy:()=>I});function u(e){return Object.fromEntries(Object.entries(e??{}).filter(([t,p])=>p.type=="static").map(([t,p])=>[t,p.value]))}async function l(e){try{let t=new TextEncoder().encode(e),p=await crypto.subtle.digest("SHA-256",t);return Array.from(new Uint8Array(p)).map(n=>n.toString(16).padStart(2,"0")).join("")}catch{let{Sha256:t}=await import("@aws-crypto/sha256-js"),p=new t;return p.update(e??""),Array.from(await p.digest()).map(n=>n.toString(16).padStart(2,"0")).join("")}}function y(e){return e?.type=="runnableByPath"||e?.type=="path"}function s(e){return e?.type=="runnableByName"||e?.type=="inline"}async function I(e,t){let p=(await Promise.all(Object.entries(e).map(async([n,o])=>await _(n,o,o?.fields??{})))).filter(n=>n!=null),a=Object.fromEntries(p);return{...t,triggerables_v2:a}}function d(e,t){let p={};typeof e.delete_after_secs=="number"&&e.delete_after_secs>=0&&(p.delete_after_secs=e.delete_after_secs);let a=Object.entries(t).map(([r,n])=>n.sensitive?r:void 0).filter(Boolean);return a.length>0&&(p.sensitive_inputs=a),e.inlineScript?.tag&&(p.tag=e.inlineScript.tag),e.viewer_data_access===!0&&s(e)&&(p.viewer_data_access=!0),p}async function _(e,t,p){let a=u(p),r=Object.entries(p).map(([n,o])=>o.allowUserResources?n:void 0).filter(Boolean);if(s(t)){let n=await l(t.inlineScript?.content);return[`${e}:rawscript/${n}`,{static_inputs:a,one_of_inputs:{},allow_user_resources:r,...d(t,p)}]}else if(y(t)){let n=t.runType!=="hubscript"?t.runType:"script";return[`${e}:${n}/${t.path}`,{static_inputs:a,one_of_inputs:{},allow_user_resources:r,...d(t,p)}]}}return R(v);})();
diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs
index 778f504c54..1a7a073732 100644
--- a/backend/windmill-api/src/jobs.rs
+++ b/backend/windmill-api/src/jobs.rs
@@ -9069,7 +9069,8 @@ async fn run_inline_preview_script(
     if let Some(job_id) = job_id {
         register_potential_assets_on_inline_execution(job_id, &w_id, &preview);
     }
-    let end_user_email = run_end_user_email(&db, &w_id, &authed, None).await?;
+    let (end_user_email, viewer_data_access) =
+        run_inline_preview_end_user(&db, &w_id, &authed).await?;
     let utils = get_worker_internal_server_inline_utils()?;
     let result = utils.run_inline_preview_script.as_ref()(RunInlinePreviewScriptFnParams {
         content: preview.content,
@@ -9081,6 +9082,7 @@ async fn run_inline_preview_script(
         permissioned_as: username_to_permissioned_as(&authed.username),
         permissioned_as_email: authed.email.clone(),
         end_user_email,
+        viewer_data_access,
         lang: preview.language,
         job_dir: "".to_string(),
         worker_name: "".to_string(),
diff --git a/backend/windmill-common/src/auth.rs b/backend/windmill-common/src/auth.rs
index 17310bdbbc..24d9c99661 100644
--- a/backend/windmill-common/src/auth.rs
+++ b/backend/windmill-common/src/auth.rs
@@ -366,6 +366,10 @@ pub struct JobPerms {
     /// token.
     #[serde(default)]
     pub job_token_scopes: Option<Vec<String>>,
+    /// Whether the job reads data as `end_user_email`: its data table connections that name no
+    /// role, and the object storage advanced permission rules it is checked against.
+    #[serde(default)]
+    pub viewer_data_access: bool,
 }
 
 impl From<JobPerms> for Authed {
@@ -659,7 +663,7 @@ pub async fn get_job_perms<'a, E: sqlx::PgExecutor<'a>>(
     sqlx::query_as!(
         JobPerms,
         "SELECT email, username, is_admin, is_operator, groups, folders, end_user_email,
-            job_token_scopes
+            job_token_scopes, viewer_data_access
         FROM job_perms WHERE job_id = $1 AND workspace_id = $2",
         job_id,
         w_id
diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs
index 75670fe874..cd65603d4f 100644
--- a/backend/windmill-common/src/jobs.rs
+++ b/backend/windmill-common/src/jobs.rs
@@ -476,6 +476,9 @@ pub struct RunInlinePreviewScriptFnParams {
     pub permissioned_as_email: String,
     /// The caller's recorded end user: `job_perms.end_user_email` of the calling job.
     pub end_user_email: Option<String>,
+    /// `job_perms.viewer_data_access` of the calling job: the code is the caller's own, so its
+    /// data table connections are made the way the caller's are.
+    pub viewer_data_access: bool,
     pub base_internal_url: String,
     pub worker_name: String,
     pub conn: crate::worker::Connection,
diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs
index 30eb2403b1..b8e4ce50e6 100644
--- a/backend/windmill-common/src/worker.rs
+++ b/backend/windmill-common/src/worker.rs
@@ -814,7 +814,8 @@ fn format_pull_query(peek: String) -> String {
             j.timeout, j.flow_step_id, j.cache_ttl, q.cache_ignore_s3_path, q.runnable_settings_handle, j.priority, j.raw_code, j.raw_lock, j.raw_flow,
             j.script_entrypoint_override, j.preprocessed, p.job_token_scopes, COALESCE(pj.runnable_path, j.args->>'_FLOW_PATH') as parent_runnable_path,
             COALESCE(p.email, j.permissioned_as_email) as permissioned_as_email, p.username as permissioned_as_username, p.is_admin as permissioned_as_is_admin,
-            p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders, p.end_user_email as permissioned_as_end_user_email
+            p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders, p.end_user_email as permissioned_as_end_user_email,
+            p.viewer_data_access
         FROM q, j
             LEFT JOIN v2_job_status f USING (id)
             LEFT JOIN job_perms p ON p.job_id = j.id
diff --git a/backend/windmill-common/src/workspaces.rs b/backend/windmill-common/src/workspaces.rs
index 3653a224f1..444a340c3f 100644
--- a/backend/windmill-common/src/workspaces.rs
+++ b/backend/windmill-common/src/workspaces.rs
@@ -2162,19 +2162,38 @@ pub enum DatatableAccess<'a> {
     PermissionedAs {
         permissioned_as: &'a str,
         email: &'a str,
-        /// The job's viewer: MUST be what the server recorded in `job_perms.end_user_email`,
-        /// never an argument or anything else the job or its caller can set.
-        end_user_email: Option<&'a str>,
+        connect_as: DatatableConnectAs<'a>,
     },
-    /// A job identified by id; its owner is read from `v2_job` and its viewer from `job_perms`.
-    /// For agent workers and anything else that authenticates as infrastructure rather than as
-    /// the job's user.
+    /// A job identified by id; its owner is read from `v2_job` and how it connects from
+    /// `job_perms`. 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,
 }
 
+/// Whose roles a job's connection that names no role holds.
+#[derive(Clone, Copy, Debug)]
+pub enum DatatableConnectAs<'a> {
+    /// The job's own identity, `permissioned_as`.
+    Job,
+    /// The job's viewer, for a job whose `job_perms.viewer_data_access` is set. MUST be what
+    /// the server recorded in `job_perms.end_user_email`, never anything the job or its caller
+    /// can set; `None` when it recorded none, which refuses the connection.
+    Viewer(Option<&'a str>),
+}
+
+impl<'a> DatatableConnectAs<'a> {
+    pub fn from_job_perms(as_viewer: bool, end_user_email: Option<&'a str>) -> Self {
+        if as_viewer {
+            Self::Viewer(end_user_email)
+        } else {
+            Self::Job
+        }
+    }
+}
+
 /// 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(
@@ -2218,11 +2237,11 @@ pub async fn can_use_datatable_role_in_governing_workspace(
 /// back to admin.
 ///
 /// A default role other than `admin` is a floor rather than the whole connection: a resolution
-/// naming no role holds it when the job's own identity may use it, plus the roles whose tenant
-/// lists name the job's viewer, as one login that is a member of each. With no role at all it is
-/// refused. Being an admin adds none of the viewer's and `admin` is never among them, so a viewer
-/// cannot be handed the connection that owns the tables and bypasses row-level security. A named
-/// role is that role alone.
+/// naming no role holds it when its identity may use it, plus the roles whose tenant lists name
+/// that identity, as one login that is a member of each. The identity is the job's own unless
+/// it connects as its viewer ([`DatatableConnectAs`]). With no role at all it is refused. Being
+/// an admin adds no role and `admin` is never among them, so nobody is handed the connection that
+/// owns the tables and bypasses row-level security. A named role is that role alone.
 pub async fn get_datatable_resource_from_db(
     db: &DB,
     w_id: &str,
diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs
index 68749d2b90..7c4cce6262 100644
--- a/backend/windmill-queue/src/jobs.rs
+++ b/backend/windmill-queue/src/jobs.rs
@@ -3732,6 +3732,9 @@ pub struct MiniPulledJob {
     pub trigger_kind: Option<TriggerKindLabel>,
     pub visible_to_owner: bool,
     pub permissioned_as_end_user_email: Option<String>,
+    /// `job_perms.viewer_data_access`: `None` where the job was read without it.
+    #[serde(default)]
+    pub viewer_data_access: Option<bool>,
     pub runnable_settings_handle: Option<i64>,
     /// The job's effective token scopes (`job_perms.job_token_scopes`), minted into its token.
     /// No `sqlx(default)`: a query that forgets the column must fail rather than mint an
@@ -3786,6 +3789,7 @@ impl MiniPulledJob {
             trigger_kind: None,
             visible_to_owner: false,
             permissioned_as_end_user_email: None,
+            viewer_data_access: None,
             runnable_settings_handle: None,
             concurrent_limit: None,
             concurrency_time_window_s: None,
@@ -4103,6 +4107,7 @@ impl PulledJob {
                 folders,
                 end_user_email: self.job.permissioned_as_end_user_email.clone(),
                 job_token_scopes: self.job.job_token_scopes.clone(),
+                viewer_data_access: self.job.viewer_data_access.unwrap_or(false),
             }),
             _ => None,
         };
@@ -4210,6 +4215,7 @@ pub async fn get_mini_pulled_job<'c>(
         trigger_kind as \"trigger_kind: TriggerKindLabel\",
         visible_to_owner,
         NULL as permissioned_as_end_user_email,
+        NULL::boolean as viewer_data_access,
         job_perms.job_token_scopes
         FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id
         LEFT JOIN job_perms ON job_perms.job_id = v2_job_queue.id WHERE v2_job_queue.id = $1",
@@ -8741,7 +8747,8 @@ pub async fn get_same_worker_job(
                     v2_job.raw_flow,
                     COALESCE(pj.runnable_path, v2_job.args->>'_FLOW_PATH') as parent_runnable_path,
                     p.email as permissioned_as_email, p.username as permissioned_as_username, p.is_admin as permissioned_as_is_admin,
-                    p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders, p.end_user_email as permissioned_as_end_user_email
+                    p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders, p.end_user_email as permissioned_as_end_user_email,
+                    p.viewer_data_access
                     FROM v2_job_queue
                     INNER JOIN v2_job ON v2_job.id = v2_job_queue.id
                     LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id
diff --git a/backend/windmill-queue/tests/debounce_test.rs b/backend/windmill-queue/tests/debounce_test.rs
index 14f7ec2dd9..692a1108b3 100644
--- a/backend/windmill-queue/tests/debounce_test.rs
+++ b/backend/windmill-queue/tests/debounce_test.rs
@@ -3457,6 +3457,7 @@ mod debounce {
             trigger_kind: None,
             visible_to_owner: false,
             permissioned_as_end_user_email: None,
+            viewer_data_access: None,
             runnable_settings_handle: rs_handle,
             job_token_scopes: None,
         };
@@ -3699,6 +3700,7 @@ mod debounce {
             trigger_kind: None,
             visible_to_owner: false,
             permissioned_as_end_user_email: None,
+            viewer_data_access: None,
             runnable_settings_handle: rs_handle,
             job_token_scopes: None,
         };
diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs
index 6088fb4960..50af562062 100644
--- a/backend/windmill-worker/src/common.rs
+++ b/backend/windmill-worker/src/common.rs
@@ -1726,20 +1726,26 @@ pub async fn cached_result_path(
         hasher.update(b"wac_step:");
         hasher.update(step_key.as_bytes());
     }
-    // What a run returns can depend on who it ran for (`WM_END_USER_EMAIL`, a data table used
-    // as the viewer), so a result cached for one viewer is never served to another. Read from
-    // `job_perms`: a flow job re-fetched between steps does not carry it.
-    let end_user: Option<Option<String>> = sqlx::query_scalar(
-        "SELECT end_user_email FROM job_perms WHERE job_id = $1 AND workspace_id = $2",
+    // What a run returns depends on whose roles its data table connections hold, its own
+    // identity's or its viewer's, and `WM_END_USER_EMAIL` names the viewer: a result cached for
+    // one of either, or in one of the two modes, is never served to another. The arguments follow
+    // undelimited, so every field here is fixed-width or length-prefixed and always present: an
+    // optional marker would read the same as an argument whose name starts with it.
+    hasher.update(b"permissioned_as:");
+    hasher.update((job.permissioned_as.len() as u64).to_le_bytes());
+    hasher.update(job.permissioned_as.as_bytes());
+    // Read from `job_perms`: a flow job re-fetched between steps does not carry them.
+    let perms: Option<(Option<String>, bool)> = sqlx::query_as(
+        "SELECT end_user_email, viewer_data_access FROM job_perms
+         WHERE job_id = $1 AND workspace_id = $2",
     )
     .bind(job.id)
     .bind(&job.workspace_id)
     .fetch_optional(db)
     .await?;
-    // Length-prefixed: the arguments follow undelimited, and the end of one viewer's address
-    // must not read as the start of another run's first argument name.
-    if let Some(end_user) = end_user.flatten() {
-        hasher.update(b"end_user:");
+    let (end_user, as_viewer) = perms.unwrap_or_default();
+    hasher.update([as_viewer as u8, end_user.is_some() as u8]);
+    if let Some(end_user) = end_user {
         hasher.update((end_user.len() as u64).to_le_bytes());
         hasher.update(end_user.as_bytes());
     }
@@ -2487,46 +2493,58 @@ mod tests {
             trigger_kind: None,
             visible_to_owner: false,
             permissioned_as_end_user_email: None,
+            viewer_data_access: None,
             runnable_settings_handle: None,
             job_token_scopes: None,
         }
     }
 
     // A cache hit returns before the job runs, so the key is all that keeps a result computed
-    // for one viewer (a data table used as the viewer, `WM_END_USER_EMAIL`) from another.
+    // for one identity or viewer (whose data table roles it read with, `WM_END_USER_EMAIL`)
+    // from another.
     #[sqlx::test(migrations = "../migrations")]
-    async fn a_cached_result_is_keyed_by_the_recorded_viewer(pool: sqlx::PgPool) {
+    async fn a_cached_result_is_keyed_by_the_identity_and_the_recorded_viewer(pool: sqlx::PgPool) {
         let client = test_client();
 
         // The last two differ only in where the viewer's address ends and the first argument
         // name starts.
         let mut paths = vec![];
-        for (viewer, arg) in [
-            (Some("a@windmill.dev"), "mode"),
-            (Some("a@windmill.dev"), "mode"),
-            (Some("b@windmill.dev"), "mode"),
-            (None, "mode"),
-            (Some("a@windmill.co"), "mode"),
-            (Some("a@windmill.com"), "ode"),
+        for (owner, viewer, as_viewer, arg) in [
+            ("u/publisher", Some("a@windmill.dev"), false, "mode"),
+            ("u/publisher", Some("a@windmill.dev"), false, "mode"),
+            ("u/publisher", Some("b@windmill.dev"), false, "mode"),
+            ("u/publisher", None, false, "mode"),
+            ("u/publisher", Some("a@windmill.co"), false, "mode"),
+            ("u/publisher", Some("a@windmill.com"), false, "ode"),
+            ("u/other", None, false, "mode"),
+            ("u/publisher", Some("a@windmill.dev"), true, "mode"),
+            ("u/publisher", None, false, "viewer_data_access;mode"),
+            ("u/publisher", None, true, "mode"),
         ] {
             let args = HashMap::from([(arg.to_string(), to_raw_value(&"all"))]);
             let job = MiniPulledJob {
                 id: uuid::Uuid::new_v4(),
                 args: Some(sqlx::types::Json(args)),
+                permissioned_as: owner.to_string(),
                 ..test_job()
             };
             sqlx::query(
                 "INSERT INTO job_perms (job_id, email, username, is_admin, is_operator,
-                     workspace_id, groups, folders, end_user_email)
+                     workspace_id, groups, folders, end_user_email, viewer_data_access)
                  VALUES ($1, 'publisher@windmill.dev', 'publisher', false, false, 'test', '{}',
-                     '{}', $2)",
+                     '{}', $2, $3)",
             )
             .bind(job.id)
             .bind(viewer)
+            .bind(as_viewer)
             .execute(&pool)
             .await
             .unwrap();
-            paths.push(cached_result_path(&pool, &client, &job, None).await.unwrap());
+            paths.push(
+                cached_result_path(&pool, &client, &job, None)
+                    .await
+                    .unwrap(),
+            );
         }
         assert_eq!(
             paths[0], paths[1],
@@ -2535,7 +2553,19 @@ mod tests {
         assert_ne!(paths[0], paths[2], "two viewers share a cached result");
         assert_ne!(paths[0], paths[3]);
         assert_ne!(paths[2], paths[3]);
-        assert_ne!(paths[4], paths[5], "an address runs into the argument names");
+        assert_ne!(
+            paths[4], paths[5],
+            "an address runs into the argument names"
+        );
+        assert_ne!(paths[3], paths[6], "two owners share a cached result");
+        assert_ne!(
+            paths[0], paths[7],
+            "a result read with the app's roles is served once the runnable connects as its viewer"
+        );
+        assert_ne!(
+            paths[8], paths[9],
+            "an argument name reads as the mode of another run"
+        );
     }
 
     #[tokio::test]
diff --git a/backend/windmill-worker/src/duckdb_executor.rs b/backend/windmill-worker/src/duckdb_executor.rs
index 58e47b9332..513d30d181 100644
--- a/backend/windmill-worker/src/duckdb_executor.rs
+++ b/backend/windmill-worker/src/duckdb_executor.rs
@@ -14,7 +14,7 @@ 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, get_ducklake_from_db_unchecked,
-    strip_fork_reserved_attach_args, DatatableAccess, DucklakeCatalogResourceType,
+    strip_fork_reserved_attach_args, DatatableAccess, DatatableConnectAs, DucklakeCatalogResourceType,
 };
 use windmill_common::PgDatabase;
 use windmill_object_store::S3_PROXY_LAST_ERRORS_CACHE;
@@ -2758,7 +2758,10 @@ async fn transform_attach_datatable(
                 DatatableAccess::PermissionedAs {
                     permissioned_as: &job.permissioned_as,
                     email: &job.permissioned_as_email,
-                    end_user_email: job.permissioned_as_end_user_email.as_deref(),
+                    connect_as: DatatableConnectAs::from_job_perms(
+                        job.viewer_data_access.unwrap_or(false),
+                        job.permissioned_as_end_user_email.as_deref(),
+                    ),
                 },
             )
             .await?
diff --git a/backend/windmill-worker/src/pg_executor.rs b/backend/windmill-worker/src/pg_executor.rs
index 6309ba8c95..12a3665c27 100644
--- a/backend/windmill-worker/src/pg_executor.rs
+++ b/backend/windmill-worker/src/pg_executor.rs
@@ -26,7 +26,7 @@ use windmill_common::worker::{
     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,
+    get_datatable_resource_from_db, parse_datatable_ref, parse_datatable_ref_for, DatatableAccess, DatatableConnectAs,
 };
 use windmill_common::{PgDatabase, PrepareQueryColumnInfo, PrepareQueryResult, DB};
 use windmill_parser::{Arg, Typ};
@@ -1229,7 +1229,10 @@ pub async fn do_postgresql(
                             DatatableAccess::PermissionedAs {
                                 permissioned_as: &job.permissioned_as,
                                 email: &job.permissioned_as_email,
-                                end_user_email: job.permissioned_as_end_user_email.as_deref(),
+                                connect_as: DatatableConnectAs::from_job_perms(
+                                    job.viewer_data_access.unwrap_or(false),
+                                    job.permissioned_as_end_user_email.as_deref(),
+                                ),
                             },
                         )
                         .await?
diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs
index cb4b378547..a55f06a191 100644
--- a/backend/windmill-worker/src/worker.rs
+++ b/backend/windmill-worker/src/worker.rs
@@ -7266,6 +7266,7 @@ pub fn init_worker_internal_server_inline_utils(
                 Some(params.lang),
             );
             job.permissioned_as_end_user_email = params.end_user_email;
+            job.viewer_data_access = Some(params.viewer_data_access);
             Box::pin(async move {
                 let mut mem_peak: i32 = -1;
                 let mut canceled_by: Option<CanceledBy> = None;
diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs
index 4f67780f18..a257ca51fa 100644
--- a/backend/windmill-worker/src/worker_flow.rs
+++ b/backend/windmill-worker/src/worker_flow.rs
@@ -2440,6 +2440,7 @@ async fn advance_flow_status(
             v2_job.trigger_kind as \"trigger_kind: TriggerKindLabel\",
             v2_job.visible_to_owner,
             NULL as permissioned_as_end_user_email,
+            NULL::boolean as viewer_data_access,
             job_perms.job_token_scopes",
         flow,
         &step_path as &[&str],
diff --git a/cli/src/commands/app/raw_apps.ts b/cli/src/commands/app/raw_apps.ts
index 4c3d8d5215..e77e1aed45 100644
--- a/cli/src/commands/app/raw_apps.ts
+++ b/cli/src/commands/app/raw_apps.ts
@@ -569,6 +569,26 @@ export async function pushRawApp(
   );
 }
 
+/**
+ * `@windmill-labs/shared-utils` up to 1.0.13 derives the policy without `viewer_data_access`,
+ * which would silently deploy an inline runnable that asks to connect as its viewer as one
+ * connecting as the app. The server honors the flag on inline runnables only.
+ */
+export function carryViewerDataAccess(
+  runnables: Record<string, any> | undefined,
+  policy: Policy | undefined,
+) {
+  const triggerables = policy?.triggerables_v2 as Record<string, any> | undefined;
+  if (!runnables || !triggerables) return;
+  for (const [id, runnable] of Object.entries(runnables)) {
+    if (runnable?.viewer_data_access !== true) continue;
+    if (runnable.type !== "inline" && runnable.type !== "runnableByName") continue;
+    for (const [key, triggerable] of Object.entries(triggerables)) {
+      if (key.startsWith(`${id}:rawscript/`)) triggerable.viewer_data_access = true;
+    }
+  }
+}
+
 export async function generatingPolicy(
   app: any,
   path: string,
@@ -579,6 +599,7 @@ export async function generatingPolicy(
   log.info(colors.gray(`Generating fresh policy for app ${path}...`));
   try {
     app.policy = await windmillUtils.updateRawAppPolicy(app.runnables, base);
+    carryViewerDataAccess(app.runnables, app.policy);
     finalizeDerivedPolicy(app.policy, executionMode, isolation, true);
   } catch (e) {
     log.error(colors.red(`Error generating policy for app ${path}: ${e}`));
diff --git a/cli/src/guidance/skills.gen.ts b/cli/src/guidance/skills.gen.ts
index 0a880b05cf..7d97196752 100644
--- a/cli/src/guidance/skills.gen.ts
+++ b/cli/src/guidance/skills.gen.ts
@@ -6522,6 +6522,8 @@ fields:
     value: "default_user"
 \`\`\`
 
+\`viewer_data_access: true\` makes an inline runnable read data tables and object storage as the person using the app instead of as the app's identity (see "Reading data as the viewer").
+
 ### Referencing existing scripts
 
 To use an existing Windmill script instead of inline code:
@@ -6558,7 +6560,7 @@ data:
 - \`<datatable>/<table>\` — Specific table in public schema
 - \`<datatable>/<schema>:<table>\` — Table in specific schema
 
-**Roles:** when a datatable is under roles, its queries run as a role, which only reaches what it was granted. \`roles\` records the role the app uses each datatable through; the app's code must pass the same role: \`wmill.datatable('main', { role: 'analyst' })\` in TypeScript, \`wmill.datatable('main', role='analyst')\` in Python. A datatable without an entry is used as its default role, if it has one, plus the roles of whoever is using the app unless that default is \`admin\`.
+**Roles:** when a datatable is under roles, its queries run as a role, which only reaches what it was granted. \`roles\` records the role the app uses each datatable through; the app's code must pass the same role: \`wmill.datatable('main', { role: 'analyst' })\` in TypeScript, \`wmill.datatable('main', role='analyst')\` in Python. A datatable without an entry is used as its default role, if it has one, plus the roles of the app's identity, or of whoever is using the app for a runnable with \`viewer_data_access: true\`, unless that default is \`admin\`.
 
 ## SQL Migrations (sql_to_apply/)
 
@@ -6853,6 +6855,11 @@ frontend_api: [jobs:run, users:read]
 
 A file with a \`sandbox\` key states both settings: \`sandbox: false\` turns isolation off, and a missing \`frontend_api\` means no scopes. A file without the key leaves the deployed app's isolation and scopes as they are, so removing \`sandbox: true\` does not unsandbox it. \`frontend_api\` needs \`sandbox: true\`; the push fails otherwise.
 
+## Reading data as the viewer
+
+A backend runnable reads data as the app's identity (the publisher in publisher mode), the same for every viewer. An inline runnable can instead read data as the person using the app: its data table connections that name no role hold that viewer's roles, and object storage permission rules (which paths may be listed, read, written) are checked against the viewer. Resources, variables and everything else stay the app's, and an anonymous viewer is refused. Turn it on only when what the runnable reads should differ per viewer: row-level security on a data table, or per-user paths in object storage. Both are Enterprise Edition features.
+Set it with \`viewer_data_access: true\` in the runnable's \`backend/<id>.yaml\`.
+
 ## Data Tables
 
 Data tables are PostgreSQL databases managed by Windmill. Backend runnables query them via the \`wmill\` client; the frontend never queries them directly. **When the app needs to store or persist data** (user data, settings, application state, records, logs), use a data table.
@@ -6864,7 +6871,7 @@ Data tables are PostgreSQL databases managed by Windmill. Backend runnables quer
 3. **No DDL inside runnables**: runnables only read and write rows (SELECT, INSERT, UPDATE, DELETE) on existing tables. Never CREATE, ALTER or DROP a table from a runnable.
 4. **Qualify table names**: an unqualified name means the \`public\` schema, so write every other table as \`schema.table\`, in table creation and queries alike. The app's \`data\` config sets the default datatable and schema its tables go in; use them consistently across runnables.
 5. **Pass the role**: when the app's \`data.roles\` gives a data table a role, every \`wmill.datatable\` call on it passes that role (\`wmill.datatable('main', { role: 'analyst' })\` in TypeScript, \`wmill.datatable('main', role='analyst')\` in Python). The role only reaches what it was granted, so a query outside it fails with \`permission denied\`.
-6. **Per-viewer rows**: a data table with no entry in \`data.roles\` is called with no role. Unless its default role is \`admin\`, the connection then holds the default role, if it has one, plus every role the person using the app is a tenant of, and is refused when that is none. Row-level security policies written as \`pg_has_role(current_user, '<role>', 'member')\` return that viewer's rows with nothing added to the runnable. A role pins the connection to that one role for every viewer, so where rows should differ per viewer give the data table no \`data.roles\` entry.
+6. **Per-viewer rows**: a data table with no entry in \`data.roles\` is called with no role. Unless its default role is \`admin\`, the connection then holds the default role, if it may use it, plus every role it is a tenant of, and is refused when that is none. Whose roles those are follows the runnable's identity (see "Reading data as the viewer"). Row-level security policies written as \`pg_has_role(current_user, '<role>', 'member')\` then return the right rows with nothing added to the code. A role pins the connection to that one role either way, so a data table whose rows should differ per viewer gets no \`data.roles\` entry. Data table roles are an Enterprise Edition feature, self-hosted only.
 
 \`wmill datatable list\` lists the workspace's data tables. Create or change tables with a migration in \`sql_to_apply/\` (see "SQL Migrations" above), then add them to \`data.tables\` in \`raw_app.yaml\`.
 
diff --git a/docs/datatable-viewer-roles.md b/docs/datatable-viewer-roles.md
index 9855e4f97a..89ae3d9d50 100644
--- a/docs/datatable-viewer-roles.md
+++ b/docs/datatable-viewer-roles.md
@@ -1,20 +1,23 @@
-# Data table connections and the viewer's roles
+# Data table connections and whose roles they hold
 
 A data table under roles (Enterprise Edition) resolves a connection that names a role to that
-role alone. A connection that names none is resolved from the data table's default role:
+role alone. A connection that names none is resolved from the data table's default role and the
+roles of one **identity**:
 
 | Default role | A connection naming no role holds |
 |---|---|
-| `admin` (until changed) | `admin`, as before viewer roles existed |
-| another role | that role when the job's own identity may use it, plus **every role the job's viewer is a tenant of** |
-| none (`NO_DEFAULT_DATATABLE_ROLE`) | the viewer's roles only |
+| `admin` (until changed) | `admin`, as before identity roles existed |
+| another role | that role when the identity may use it, plus **every role the identity is a tenant of** |
+| none (`NO_DEFAULT_DATATABLE_ROLE`) | the identity's roles only |
 
-A connection left with no role at all is refused. Postgres row-level security can then answer
-per viewer from a backend runnable that runs as the app's publisher, with nothing written in the
-runnable.
+The identity is the job's own, `permissioned_as`, like everything else a job reaches: the
+publisher for a publisher-mode app, the owner of a schedule or trigger, whoever ran a script. It
+is the job's **viewer** only for an inline runnable of a full-code app whose policy entry sets
+`viewer_data_access` (`DatatableConnectAs::Viewer`). Postgres row-level security then answers per
+viewer from a runnable that runs as the publisher, with nothing written in the code.
 
-Policies test membership, which is true for a role's own login too, so one policy serves a named
-connection and an unnamed one:
+A connection left with no role at all is refused. Policies test membership, which is true for a
+role's own login too, so one policy serves a named connection and an unnamed one:
 
 ```sql
 CREATE POLICY emea ON orders FOR SELECT
@@ -27,67 +30,75 @@ A policy names its role as text, so renaming the role breaks it.
 
 | Where | What |
 |---|---|
-| `windmill-common/src/workspaces.rs` | `DatatableAccess`, `get_datatable_resource_from_db` |
-| `windmill-common/src/datatable_roles_ee.rs` | `resolve_datatable_role_connection`, `viewer_role_ids`, `ensure_role_set_login`, `retire_role_set_logins_of`, `sweep_unused_role_set_logins` |
-| `windmill-api-jobs/src/execution.rs` | `run_end_user_email`, where a run's viewer is recorded |
+| `windmill-common/src/workspaces.rs` | `DatatableAccess`, `DatatableConnectAs`, `get_datatable_resource_from_db` |
+| `windmill-common/src/datatable_roles_ee.rs` | `resolve_datatable_role_connection`, `resolve_unnamed_login`, `governing_identity`, `ensure_role_set_login`, `retire_role_set_logins_of`, `sweep_unused_role_set_logins` |
+| `windmill-api/src/apps.rs` | `execute_component`, which sets `job_perms.viewer_data_access` for a flagged inline runnable |
+| `windmill-api-jobs/src/execution.rs` | `run_end_user_email`, where a run's viewer is recorded, and `run_inline_preview_end_user` |
+| `frontend/.../raw_apps/ViewerDataAccessPopup.svelte` | the runnable setting, shown when the code reads a data table or object storage |
 | `frontend/.../workspaceSettings/DataTablePermissionsButton.svelte` | the roles drawer, which warns while the default is `admin` |
 
 ## What holds it together
 
 - **A default of `admin` is not a floor.** `admin` owns the tables and bypasses row-level
   security, so nothing could be added to it, and it is the default until someone changes it.
-  Such a data table resolves exactly as it did before viewer roles existed. Nothing ever sets a
+  Such a data table resolves exactly as it did before identity roles existed. Nothing ever sets a
   default back to `admin`: removing or deleting the default role leaves none.
-- **The default role is tenant-checked against the job's own identity**, the publisher for a
-  publisher-mode app, and held only when that passes. The viewer's roles are added either way.
+- **One identity decides both the default and the added roles.** The default is held when that
+  identity may use it, and the roles added are those whose tenants name that same identity. A
+  job never mixes its own identity's default with its viewer's roles.
+- **Connecting as the viewer is the publisher's choice, made per runnable and stored in the
+  policy.** It is honored on inline runnables only, whose code is deployed with the app: a path
+  runnable is someone else's script, which would read with the viewer's roles without its author
+  choosing to. The server stamps `job_perms.viewer_data_access` in the transaction that pushes
+  the job. Nested jobs do not inherit it; an inline preview run with the job's token
+  (`run_inline/preview`, what `wmill.datatable()` uses) does, since that code is the job's own.
+- **A job that connects as its viewer and has none is refused**, never resolved as the job: that
+  would undo what the runnable asked for. An anonymous app viewer is the usual case.
 - **No role at all is a refusal, never an empty answer.** There is no login that would read
-  nothing: what a role sees is up to each table's grants and policies. A job without a recorded
-  viewer (anonymous app viewer, schedule, trigger, a tool job started by an AI agent step), or
-  whose viewer is not a member of the governing workspace, therefore needs a default it may use
-  or a named role.
+  nothing: what a role sees is up to each table's grants and policies.
 - **A set login owns what it creates.** Postgres gives a new object to `current_user`, and
   default privileges go by the creating role, so a table created over a connection that holds
-  several roles belongs to that set's login: no other role or viewer reaches it, and it passes to
-  `custom_instance_user` when the login is swept. Unnamed connections are for rows. To create an
-  object, name a role or run `SET ROLE <role>` first, which a set login may do for any role it
+  several roles belongs to that set's login: no other role or identity reaches it, and it passes
+  to `custom_instance_user` when the login is swept. Unnamed connections are for rows. To create
+  an object, name a role or run `SET ROLE <role>` first, which a set login may do for any role it
   holds. Migrations are not affected: one that declares no role runs as `admin`.
 - **The viewer is `job_perms.end_user_email`, and nothing else.** The server records it when the
   run starts: the session behind an app request, the caller of a direct run or preview, and for a
   job started with another job's token, that job's viewer. No argument reaches it.
-- **A cached result belongs to one viewer.** `cached_result_path` hashes the run's recorded viewer
-  into the key, since a cache hit returns before any connection is resolved. Runs with no viewer
-  recorded keep sharing one entry.
-- **Only a tenant list naming the viewer counts.** A workspace admin may *use* every role,
-  `admin` included. The viewer's roles are those whose tenants name them (user, group, folder,
-  `*`), and never `admin`.
+- **A cached result belongs to one identity, one viewer and one mode.** `cached_result_path`
+  hashes the job's `permissioned_as`, its recorded viewer and `viewer_data_access` into the key,
+  since a cache hit returns before any connection is resolved. Turning the setting on changes no
+  code, so without the mode a result read with the app's roles would be served to the viewer.
+- **Only a tenant list naming the identity counts.** A workspace admin may *use* every role,
+  `admin` included. The roles added are those whose tenants name the identity (user, group,
+  folder, `*`), and never `admin`.
 - **One role is that role's login. Several are a login per distinct set.** `wm_set_<hash>` is a
   member of exactly those roles, with `INHERIT`; the hash is over their catalog ids, so a rename
   moves nothing and a change in someone's groups resolves them to another login instead of
   editing one. Its privileges are its memberships, so `SET ROLE` only narrows and `RESET ROLE`
   comes back to it. The prefix is reserved in `validate_role_name`.
-- **Resolving stays cheap where no viewer could add anything.** The viewer is only looked up when
-  some role other than the default and `admin` has tenants on that data table.
 - **A `datatable_role_set` row with a password names a login that can authenticate with it.** The
   row is written after the login exists and removed, in its own commit, before the login is
   dropped or made `NOLOGIN`. A login without a row is taken back with a new password on next use,
   and so is one whose row has an empty password: that is what a sweep leaves when it could
   neither drop a login nor let it back in, so the next sweep still finds it.
 - **Set logins are bounded by use, not by the number of role combinations.** One exists only for
-  a set some viewer actually connected with, so there are at most as many as (viewer, data table)
-  pairs. `last_used_at` is recorded at most once a day, and `sweep_unused_role_set_logins`, run
-  from the monitor loop, drops those unused for 30 days; a dropped one is made again on demand.
-  The sweep goes by this instance's rows, never by the cluster's `wm_set_*` roles: a cluster can
-  serve several Windmill instances.
+  a set some identity actually connected with, so there are at most as many as (identity, data
+  table) pairs. `last_used_at` is recorded at most once a day, and `sweep_unused_role_set_logins`,
+  run from the monitor loop, drops those unused for 30 days; a dropped one is made again on
+  demand. The sweep goes by this instance's rows, never by the cluster's `wm_set_*` roles: a
+  cluster can serve several Windmill instances.
 - **Disabling or dropping a role retires every set login holding it.** `NOLOGIN` on a role does
   not reach a login that is a member of it.
-- **A worker or server that predates this connects as the default role alone**, so a mixed fleet
-  shows a viewer fewer rows, never more. It refuses a data table with no default role, whose
-  marker it reads as a role the data table lacks.
+- **Mixed fleets.** A worker or server older than v1.831 connects as the default role alone, and
+  refuses a data table with no default role, whose marker it reads as a role the data table
+  lacks. One from v1.831 or v1.832 adds the recorded viewer's roles to every job that has one,
+  flagged or not, and the job's own roles to none.
 
 ## Trust
 
-A job that names no role runs its author's code with the viewer's reach into the data table, the
-same trust as an app in viewer execution mode or a script someone else wrote and you run. Whoever
-can deploy a publisher-mode app, or a script that others run, can therefore read through the
-non-`admin` roles of the people who use it. Naming a role in the code is how an author pins the
-connection to one role instead.
+A job that names no role reads with its own identity's roles, the same trust as the resources and
+variables it reaches: every viewer of a publisher-mode app sees what the publisher's roles see.
+An inline runnable that connects as its viewer runs its publisher's code with the viewer's reach
+into the data table, the same trust as an app in viewer execution mode. Naming a role in the code
+pins the connection to that one role either way.
diff --git a/frontend/src/lib/components/apps/editor/commonAppUtils.ts b/frontend/src/lib/components/apps/editor/commonAppUtils.ts
index 564a94e9b1..b384fafce5 100644
--- a/frontend/src/lib/components/apps/editor/commonAppUtils.ts
+++ b/frontend/src/lib/components/apps/editor/commonAppUtils.ts
@@ -17,6 +17,8 @@ export type TriggerableV2 = {
 	delete_after_secs?: number
 	sensitive_inputs?: string[]
 	tag?: string
+	/** Data table connections naming no role are made as the app's viewer. Inline runnables only. */
+	viewer_data_access?: boolean
 }
 
 export async function hash(message) {
diff --git a/frontend/src/lib/components/copilot/chat/app/core.ts b/frontend/src/lib/components/copilot/chat/app/core.ts
index b013c58ebc..166df65a16 100644
--- a/frontend/src/lib/components/copilot/chat/app/core.ts
+++ b/frontend/src/lib/components/copilot/chat/app/core.ts
@@ -37,6 +37,8 @@ export interface BackendRunnable {
 	inlineScript?: InlineScript
 	// For workspace scripts/flows and hub scripts
 	path?: string
+	// Inline only: data table connections naming no role hold the viewer's roles
+	viewerDataAccess?: boolean
 }
 
 export interface AppFiles {
@@ -327,6 +329,12 @@ const getSetBackendRunnableSchema = memo(() =>
 			.optional()
 			.describe(
 				'Required when type is "script", "flow", or "hubscript". The fully qualified path to the workspace script/flow or hub script.'
+			),
+		viewerDataAccess: z
+			.boolean()
+			.optional()
+			.describe(
+				"Inline only. When true, the runnable reads data as the person viewing the app instead of as the app's identity: data table connections that name no role hold the viewer's roles, and object storage permission rules are checked against the viewer, so row-level security and per-user storage paths answer per viewer. Resources and variables stay the app's. Anonymous viewers are refused. Omit to keep the current setting."
 			)
 	})
 )
@@ -710,7 +718,8 @@ export const getAppTools = memo((): Tool<AppAIChatHelpers>[] => [
 				...args,
 				staticInputs: args.staticInputs === null ? undefined : args.staticInputs,
 				inlineScript: args.inlineScript === null ? undefined : args.inlineScript,
-				path: args.path === null ? undefined : args.path
+				path: args.path === null ? undefined : args.path,
+				viewerDataAccess: args.viewerDataAccess === null ? undefined : args.viewerDataAccess
 			}
 
 			const parsedArgs = getSetBackendRunnableSchema().parse(cleanedArgs)
@@ -723,7 +732,10 @@ export const getAppTools = memo((): Tool<AppAIChatHelpers>[] => [
 				type: parsedArgs.type,
 				...(parsedArgs.staticInputs && { staticInputs: parsedArgs.staticInputs }),
 				...(parsedArgs.inlineScript && { inlineScript: parsedArgs.inlineScript }),
-				...(parsedArgs.path && { path: parsedArgs.path })
+				...(parsedArgs.path && { path: parsedArgs.path }),
+				...(parsedArgs.viewerDataAccess !== undefined && {
+					viewerDataAccess: parsedArgs.viewerDataAccess
+				})
 			}
 
 			toolCallbacks.setToolStatus(toolId, {
diff --git a/frontend/src/lib/components/copilot/chat/global/core.ts b/frontend/src/lib/components/copilot/chat/global/core.ts
index 506126529b..c23161c215 100644
--- a/frontend/src/lib/components/copilot/chat/global/core.ts
+++ b/frontend/src/lib/components/copilot/chat/global/core.ts
@@ -1054,6 +1054,12 @@ const backendRunnableSchema = z
 			.optional()
 			.describe(
 				'Required when type is "script", "flow", or "hubscript". Workspace path of the runnable.'
+			),
+		viewerDataAccess: z
+			.boolean()
+			.optional()
+			.describe(
+				"Inline only. When true, the runnable reads data as the person viewing the app instead of as the app's identity: data table connections that name no role hold the viewer's roles, and object storage permission rules are checked against the viewer, so row-level security and per-user storage paths answer per viewer. Resources and variables stay the app's. Anonymous viewers are refused. Omit to keep the current setting."
 			)
 	})
 	.describe(
@@ -1854,6 +1860,7 @@ function convertPersistedToBackendRunnable(
 				content: persisted.inlineScript.content ?? ''
 			}
 		}
+		if (persisted.viewer_data_access === true) out.viewerDataAccess = true
 	} else if (persisted.type === 'path' || persisted.type === 'runnableByPath') {
 		if (persisted.runType === 'flow') out.type = 'flow'
 		else if (persisted.runType === 'hubscript') out.type = 'hubscript'
@@ -1891,7 +1898,14 @@ function buildPersistedRunnable(
 	// `type` (isRunnableByName / isRunnableByPath), so a leftover does not change which code
 	// runs — it contradicts it: a runnable reported back as inline still names a flow in
 	// `path`, and that is what the model reads on its next turn.
-	const { runType: _runType, path: _path, inlineScript: _inlineScript, ...carried } = existing ?? {}
+	const {
+		runType: _runType,
+		path: _path,
+		inlineScript: _inlineScript,
+		viewer_data_access: carriedAsViewer,
+		...carried
+	}: Partial<PersistedRunnable> = existing ?? {}
+	const viewerDataAccess = input.viewerDataAccess ?? carriedAsViewer === true
 
 	// `schema` describes the item a path runnable points at, so it only survives while that
 	// target is unchanged. Carried across a retarget it types `backend.<key>(args)` from the
@@ -1913,6 +1927,7 @@ function buildPersistedRunnable(
 				content: input.inlineScript.content,
 				language: input.inlineScript.language
 			},
+			...(viewerDataAccess && { viewer_data_access: true }),
 			fields
 		}
 	}
diff --git a/frontend/src/lib/components/raw_apps/RawAppEditor.svelte b/frontend/src/lib/components/raw_apps/RawAppEditor.svelte
index b26e5df15f..454030009b 100644
--- a/frontend/src/lib/components/raw_apps/RawAppEditor.svelte
+++ b/frontend/src/lib/components/raw_apps/RawAppEditor.svelte
@@ -897,6 +897,7 @@ addEventListener('message', function (e) {
 					content: runnable.inlineScript.content ?? ''
 				}
 			}
+			if (runnable.viewer_data_access) backendRunnable.viewerDataAccess = true
 		} else if (isRunnableByPath(runnable)) {
 			// Determine type based on runType
 			if (runnable.runType === 'flow') {
@@ -1200,6 +1201,9 @@ addEventListener('message', function (e) {
 							content: runnable.inlineScript.content,
 							language: runnable.inlineScript.language
 						},
+						...((runnable.viewerDataAccess ?? runnables[key]?.viewer_data_access) && {
+							viewer_data_access: true
+						}),
 						fields: runnable.staticInputs
 							? Object.fromEntries(
 									Object.entries(runnable.staticInputs).map(([k, v]) => [
diff --git a/frontend/src/lib/components/raw_apps/RawAppInlineScriptEditor.svelte b/frontend/src/lib/components/raw_apps/RawAppInlineScriptEditor.svelte
index d7ae7b2245..170c3a1b5b 100644
--- a/frontend/src/lib/components/raw_apps/RawAppInlineScriptEditor.svelte
+++ b/frontend/src/lib/components/raw_apps/RawAppInlineScriptEditor.svelte
@@ -17,6 +17,7 @@
 	import CacheTtlPopup from '../apps/editor/inlineScriptsPanel/CacheTtlPopup.svelte'
 	import TagPopup from '../apps/editor/inlineScriptsPanel/TagPopup.svelte'
 	import DeleteAfterUsePopup from './DeleteAfterUsePopup.svelte'
+	import ViewerDataAccessPopup from './ViewerDataAccessPopup.svelte'
 	import { computeFields } from '../apps/editor/inlineScriptsPanel/utils'
 	import EditorBar, { EDITOR_BAR_HELPERS_INLINE_THRESHOLD } from '../EditorBar.svelte'
 	import { LanguageIcon } from '../common/languageIcons'
@@ -52,6 +53,7 @@
 		editor?: any | undefined
 		lastDeployedCode?: string | undefined
 		delete_after_secs?: number | undefined
+		viewer_data_access?: boolean | undefined
 		/** Called when code is selected in the editor */
 		onSelectionChange?: (
 			selection: {
@@ -74,9 +76,17 @@
 		editor = $bindable(undefined),
 		lastDeployedCode,
 		onSelectionChange,
-		delete_after_secs = $bindable()
+		delete_after_secs = $bindable(),
+		viewer_data_access = $bindable()
 	}: Props = $props()
 
+	// The asset parser misses a chained `wmill.datatable('x')`...``, so the source is matched too.
+	const readsData = $derived(
+		(inlineScript?.assets?.some((a) => ['datatable', 'ducklake', 's3object'].includes(a.kind)) ??
+			false) ||
+			/datatable\s*\(|datatable:\/\/|s3:\/\/|s3_?file|S3Object/i.test(inlineScript?.content ?? '')
+	)
+
 	const operatingWorkspace = useOperatingWorkspace()
 	let opWs = $derived($operatingWorkspace)
 
@@ -620,6 +630,9 @@
 				/>
 			{/if}
 			<div class="flex w-full flex-row gap-2 items-center justify-end">
+				{#if readsData || viewer_data_access}
+					<ViewerDataAccessPopup bind:viewer_data_access />
+				{/if}
 				{#if inlineScript}
 					<TagPopup bind:tag={inlineScript.tag} />
 					<CacheTtlPopup bind:cache_ttl={inlineScript.cache_ttl} />
diff --git a/frontend/src/lib/components/raw_apps/RawAppInlineScriptRunnable.svelte b/frontend/src/lib/components/raw_apps/RawAppInlineScriptRunnable.svelte
index ddbdae5bf3..73fe0852f4 100644
--- a/frontend/src/lib/components/raw_apps/RawAppInlineScriptRunnable.svelte
+++ b/frontend/src/lib/components/raw_apps/RawAppInlineScriptRunnable.svelte
@@ -43,6 +43,7 @@
 	type RunnableWithInlineScript = RunnableWithFields & {
 		inlineScript?: InlineScript & { language: ScriptLang }
 		delete_after_secs?: number
+		viewer_data_access?: boolean
 	}
 	export type Runnable = RunnableWithInlineScript | undefined
 	interface Props {
@@ -224,6 +225,7 @@
 					bind:name={runnable.name}
 					bind:fields={runnable.fields}
 					bind:delete_after_secs={runnable.delete_after_secs}
+					bind:viewer_data_access={runnable.viewer_data_access}
 					onRun={testPreview}
 					on:delete
 					path={appPath}
diff --git a/frontend/src/lib/components/raw_apps/ViewerDataAccessPopup.svelte b/frontend/src/lib/components/raw_apps/ViewerDataAccessPopup.svelte
new file mode 100644
index 0000000000..df0215c459
--- /dev/null
+++ b/frontend/src/lib/components/raw_apps/ViewerDataAccessPopup.svelte
@@ -0,0 +1,86 @@
+<script lang="ts">
+	import Toggle from '$lib/components/Toggle.svelte'
+	import { Button } from '$lib/components/common'
+	import { AppWindow, UserRound } from 'lucide-svelte'
+	import Popover from '$lib/components/meltComponents/Popover.svelte'
+	import { autoPlacement } from '@floating-ui/core'
+	import { enterpriseLicense } from '$lib/stores'
+	import { isCloudHosted } from '$lib/cloud'
+
+	interface Props {
+		viewer_data_access: boolean | undefined
+	}
+
+	let { viewer_data_access = $bindable() }: Props = $props()
+
+	// It only changes what data table roles and object storage permission rules decide, both
+	// Enterprise Edition.
+	const unavailable = $derived(
+		!$enterpriseLicense
+			? 'Data table roles and object storage permission rules are Enterprise Edition features.'
+			: undefined
+	)
+</script>
+
+<Popover
+	floatingConfig={{
+		middleware: [
+			autoPlacement({
+				allowedPlacements: ['bottom-start', 'bottom-end', 'top-start', 'top-end', 'top', 'bottom']
+			})
+		]
+	}}
+	closeButton
+	contentClasses="block text-primary p-4 max-w-sm"
+>
+	{#snippet trigger()}
+		<Button
+			nonCaptureEvent={true}
+			variant={viewer_data_access ? 'accent-secondary' : 'default'}
+			unifiedSize="sm"
+			startIcon={{ icon: viewer_data_access ? UserRound : AppWindow }}
+			iconOnly
+			title={(viewer_data_access
+				? 'Reads data tables and object storage as the viewer'
+				: 'Reads data tables and object storage as the app') +
+				(unavailable ? ' (Enterprise Edition)' : '')}
+		/>
+	{/snippet}
+	{#snippet content()}
+		<Toggle
+			checked={viewer_data_access ?? false}
+			disabled={unavailable !== undefined && !viewer_data_access}
+			eeOnly
+			on:change={() => {
+				viewer_data_access = viewer_data_access ? undefined : true
+			}}
+			options={{
+				right: 'Read data as the viewer',
+				rightTooltip:
+					'Data table roles map who may use a data table to Postgres roles, so grants and row-level security decide what a connection sees. Object storage permission rules decide which paths an identity may list, read and write.',
+				rightDocumentationLink:
+					'https://www.windmill.dev/docs/core_concepts/persistent_storage/data_tables'
+			}}
+		/>
+		<div class="mt-3 flex flex-col gap-2 text-xs text-secondary">
+			{#if unavailable}
+				<p class="text-primary">{unavailable}</p>
+			{/if}
+			<p>
+				By default this runnable reads as the identity the app runs as, the publisher in publisher
+				mode, for every viewer: a data table connection that names no role holds that identity's
+				roles, and object storage permission rules are checked against it.
+			</p>
+			<p>
+				Turn this on to read as the person viewing the app instead, so row-level security and
+				per-user storage paths answer per viewer. Anonymous viewers are refused. Resources,
+				variables and everything else stay the app's.
+			</p>
+			<p>
+				A data table connection that names a role is that role alone either way, and one whose
+				default role is admin stays admin, which bypasses row-level security.
+				{#if isCloudHosted()}Data table roles are only available on self-hosted instances.{/if}
+			</p>
+		</div>
+	{/snippet}
+</Popover>
diff --git a/frontend/src/lib/components/raw_apps/rawAppPolicy.ts b/frontend/src/lib/components/raw_apps/rawAppPolicy.ts
index ca84b39a4b..c3ddc5f1a0 100644
--- a/frontend/src/lib/components/raw_apps/rawAppPolicy.ts
+++ b/frontend/src/lib/components/raw_apps/rawAppPolicy.ts
@@ -29,14 +29,15 @@ export async function updateRawAppPolicy(
 type RunnableWithInlineScript = RunnableWithFields & {
 	inlineScript?: InlineScript & { language: ScriptLang }
 	delete_after_secs?: number
+	viewer_data_access?: boolean
 }
 export type Runnable = RunnableWithInlineScript | undefined
 
 function extraFields(
 	runnable: RunnableWithInlineScript,
 	fields: Record<string, any>
-): Partial<Pick<TriggerableV2, 'delete_after_secs' | 'sensitive_inputs' | 'tag'>> {
-	const out: Partial<Pick<TriggerableV2, 'delete_after_secs' | 'sensitive_inputs' | 'tag'>> = {}
+): Partial<TriggerableV2> {
+	const out: Partial<TriggerableV2> = {}
 	if (typeof runnable.delete_after_secs === 'number' && runnable.delete_after_secs >= 0)
 		out.delete_after_secs = runnable.delete_after_secs
 	const sensitive_inputs = Object.entries(fields)
@@ -44,6 +45,9 @@ function extraFields(
 		.filter(Boolean) as string[]
 	if (sensitive_inputs.length > 0) out.sensitive_inputs = sensitive_inputs
 	if (runnable.inlineScript?.tag) out.tag = runnable.inlineScript.tag
+	// The server honors it on inline runnables only, whose code is deployed with the app.
+	if (runnable.viewer_data_access === true && isRunnableByName(runnable))
+		out.viewer_data_access = true
 	return out
 }
 
diff --git a/frontend/src/lib/components/raw_apps/utils.ts b/frontend/src/lib/components/raw_apps/utils.ts
index 3070572785..e7b5a37384 100644
--- a/frontend/src/lib/components/raw_apps/utils.ts
+++ b/frontend/src/lib/components/raw_apps/utils.ts
@@ -10,6 +10,7 @@ import { appSourceToDraftValue, normalizeRawAppData } from './rawAppDraftValue'
 
 type RunnableWithInlineScript = RunnableWithFields & {
 	inlineScript?: InlineScript & { language: ScriptLang }
+	viewer_data_access?: boolean
 }
 export type Runnable = RunnableWithInlineScript | undefined
 
diff --git a/system_prompts/auto-generated/prompts.ts b/system_prompts/auto-generated/prompts.ts
index 54d869f727..0188c6e6d0 100644
--- a/system_prompts/auto-generated/prompts.ts
+++ b/system_prompts/auto-generated/prompts.ts
@@ -1120,6 +1120,11 @@ When a deploy widens who may open the app, tell the user in plain words.
 
 By default a deployed app's frontend runs with the viewer's own Windmill session. A sandboxed app is isolated from it: the frontend reaches Windmill only through its backend runnables, plus whatever frontend API scopes the app declares, which each viewer approves before the app runs. The scopes are \`jobs:run\`, \`jobs:read\`, \`users:read\`, \`resources:read\`, \`variables:read\`, \`flow_conversations:read\` and \`flow_conversations:write\`; grant only the ones the frontend calls.
 
+## Reading data as the viewer
+
+A backend runnable reads data as the app's identity (the publisher in publisher mode), the same for every viewer. An inline runnable can instead read data as the person using the app: its data table connections that name no role hold that viewer's roles, and object storage permission rules (which paths may be listed, read, written) are checked against the viewer. Resources, variables and everything else stay the app's, and an anonymous viewer is refused. Turn it on only when what the runnable reads should differ per viewer: row-level security on a data table, or per-user paths in object storage. Both are Enterprise Edition features.
+Set it with \`viewerDataAccess: true\` on the backend runnable.
+
 ## Data Tables
 
 Data tables are PostgreSQL databases managed by Windmill. Backend runnables query them via the \`wmill\` client; the frontend never queries them directly. **When the app needs to store or persist data** (user data, settings, application state, records, logs), use a data table.
@@ -1131,7 +1136,7 @@ Data tables are PostgreSQL databases managed by Windmill. Backend runnables quer
 3. **No DDL inside runnables**: runnables only read and write rows (SELECT, INSERT, UPDATE, DELETE) on existing tables. Never CREATE, ALTER or DROP a table from a runnable.
 4. **Qualify table names**: an unqualified name means the \`public\` schema, so write every other table as \`schema.table\`, in table creation and queries alike. The app's \`data\` config sets the default datatable and schema its tables go in; use them consistently across runnables.
 5. **Pass the role**: when the app's \`data.roles\` gives a data table a role, every \`wmill.datatable\` call on it passes that role (\`wmill.datatable('main', { role: 'analyst' })\` in TypeScript, \`wmill.datatable('main', role='analyst')\` in Python). The role only reaches what it was granted, so a query outside it fails with \`permission denied\`.
-6. **Per-viewer rows**: a data table with no entry in \`data.roles\` is called with no role. Unless its default role is \`admin\`, the connection then holds the default role, if it has one, plus every role the person using the app is a tenant of, and is refused when that is none. Row-level security policies written as \`pg_has_role(current_user, '<role>', 'member')\` return that viewer's rows with nothing added to the runnable. A role pins the connection to that one role for every viewer, so where rows should differ per viewer give the data table no \`data.roles\` entry.
+6. **Per-viewer rows**: a data table with no entry in \`data.roles\` is called with no role. Unless its default role is \`admin\`, the connection then holds the default role, if it may use it, plus every role it is a tenant of, and is refused when that is none. Whose roles those are follows the runnable's identity (see "Reading data as the viewer"). Row-level security policies written as \`pg_has_role(current_user, '<role>', 'member')\` then return the right rows with nothing added to the code. A role pins the connection to that one role either way, so a data table whose rows should differ per viewer gets no \`data.roles\` entry. Data table roles are an Enterprise Edition feature, self-hosted only.
 
 ### Querying in TypeScript (Bun/Deno)
 
diff --git a/system_prompts/auto-generated/skills/raw-app/SKILL.md b/system_prompts/auto-generated/skills/raw-app/SKILL.md
index bd21352995..c4efe0da13 100644
--- a/system_prompts/auto-generated/skills/raw-app/SKILL.md
+++ b/system_prompts/auto-generated/skills/raw-app/SKILL.md
@@ -152,6 +152,8 @@ fields:
     value: "default_user"
 ```
 
+`viewer_data_access: true` makes an inline runnable read data tables and object storage as the person using the app instead of as the app's identity (see "Reading data as the viewer").
+
 ### Referencing existing scripts
 
 To use an existing Windmill script instead of inline code:
@@ -188,7 +190,7 @@ data:
 - `<datatable>/<table>` — Specific table in public schema
 - `<datatable>/<schema>:<table>` — Table in specific schema
 
-**Roles:** when a datatable is under roles, its queries run as a role, which only reaches what it was granted. `roles` records the role the app uses each datatable through; the app's code must pass the same role: `wmill.datatable('main', { role: 'analyst' })` in TypeScript, `wmill.datatable('main', role='analyst')` in Python. A datatable without an entry is used as its default role, if it has one, plus the roles of whoever is using the app unless that default is `admin`.
+**Roles:** when a datatable is under roles, its queries run as a role, which only reaches what it was granted. `roles` records the role the app uses each datatable through; the app's code must pass the same role: `wmill.datatable('main', { role: 'analyst' })` in TypeScript, `wmill.datatable('main', role='analyst')` in Python. A datatable without an entry is used as its default role, if it has one, plus the roles of the app's identity, or of whoever is using the app for a runnable with `viewer_data_access: true`, unless that default is `admin`.
 
 ## SQL Migrations (sql_to_apply/)
 
@@ -483,6 +485,11 @@ frontend_api: [jobs:run, users:read]
 
 A file with a `sandbox` key states both settings: `sandbox: false` turns isolation off, and a missing `frontend_api` means no scopes. A file without the key leaves the deployed app's isolation and scopes as they are, so removing `sandbox: true` does not unsandbox it. `frontend_api` needs `sandbox: true`; the push fails otherwise.
 
+## Reading data as the viewer
+
+A backend runnable reads data as the app's identity (the publisher in publisher mode), the same for every viewer. An inline runnable can instead read data as the person using the app: its data table connections that name no role hold that viewer's roles, and object storage permission rules (which paths may be listed, read, written) are checked against the viewer. Resources, variables and everything else stay the app's, and an anonymous viewer is refused. Turn it on only when what the runnable reads should differ per viewer: row-level security on a data table, or per-user paths in object storage. Both are Enterprise Edition features.
+Set it with `viewer_data_access: true` in the runnable's `backend/<id>.yaml`.
+
 ## Data Tables
 
 Data tables are PostgreSQL databases managed by Windmill. Backend runnables query them via the `wmill` client; the frontend never queries them directly. **When the app needs to store or persist data** (user data, settings, application state, records, logs), use a data table.
@@ -494,7 +501,7 @@ Data tables are PostgreSQL databases managed by Windmill. Backend runnables quer
 3. **No DDL inside runnables**: runnables only read and write rows (SELECT, INSERT, UPDATE, DELETE) on existing tables. Never CREATE, ALTER or DROP a table from a runnable.
 4. **Qualify table names**: an unqualified name means the `public` schema, so write every other table as `schema.table`, in table creation and queries alike. The app's `data` config sets the default datatable and schema its tables go in; use them consistently across runnables.
 5. **Pass the role**: when the app's `data.roles` gives a data table a role, every `wmill.datatable` call on it passes that role (`wmill.datatable('main', { role: 'analyst' })` in TypeScript, `wmill.datatable('main', role='analyst')` in Python). The role only reaches what it was granted, so a query outside it fails with `permission denied`.
-6. **Per-viewer rows**: a data table with no entry in `data.roles` is called with no role. Unless its default role is `admin`, the connection then holds the default role, if it has one, plus every role the person using the app is a tenant of, and is refused when that is none. Row-level security policies written as `pg_has_role(current_user, '<role>', 'member')` return that viewer's rows with nothing added to the runnable. A role pins the connection to that one role for every viewer, so where rows should differ per viewer give the data table no `data.roles` entry.
+6. **Per-viewer rows**: a data table with no entry in `data.roles` is called with no role. Unless its default role is `admin`, the connection then holds the default role, if it may use it, plus every role it is a tenant of, and is refused when that is none. Whose roles those are follows the runnable's identity (see "Reading data as the viewer"). Row-level security policies written as `pg_has_role(current_user, '<role>', 'member')` then return the right rows with nothing added to the code. A role pins the connection to that one role either way, so a data table whose rows should differ per viewer gets no `data.roles` entry. Data table roles are an Enterprise Edition feature, self-hosted only.
 
 `wmill datatable list` lists the workspace's data tables. Create or change tables with a migration in `sql_to_apply/` (see "SQL Migrations" above), then add them to `data.tables` in `raw_app.yaml`.
 
diff --git a/system_prompts/base/raw-app-cli.md b/system_prompts/base/raw-app-cli.md
index e22de102f6..8bd1a23cf0 100644
--- a/system_prompts/base/raw-app-cli.md
+++ b/system_prompts/base/raw-app-cli.md
@@ -147,6 +147,8 @@ fields:
     value: "default_user"
 ```
 
+`viewer_data_access: true` makes an inline runnable read data tables and object storage as the person using the app instead of as the app's identity (see "Reading data as the viewer").
+
 ### Referencing existing scripts
 
 To use an existing Windmill script instead of inline code:
@@ -183,7 +185,7 @@ data:
 - `<datatable>/<table>` — Specific table in public schema
 - `<datatable>/<schema>:<table>` — Table in specific schema
 
-**Roles:** when a datatable is under roles, its queries run as a role, which only reaches what it was granted. `roles` records the role the app uses each datatable through; the app's code must pass the same role: `wmill.datatable('main', { role: 'analyst' })` in TypeScript, `wmill.datatable('main', role='analyst')` in Python. A datatable without an entry is used as its default role, if it has one, plus the roles of whoever is using the app unless that default is `admin`.
+**Roles:** when a datatable is under roles, its queries run as a role, which only reaches what it was granted. `roles` records the role the app uses each datatable through; the app's code must pass the same role: `wmill.datatable('main', { role: 'analyst' })` in TypeScript, `wmill.datatable('main', role='analyst')` in Python. A datatable without an entry is used as its default role, if it has one, plus the roles of the app's identity, or of whoever is using the app for a runnable with `viewer_data_access: true`, unless that default is `admin`.
 
 ## SQL Migrations (sql_to_apply/)
 
diff --git a/system_prompts/base/raw-app.md b/system_prompts/base/raw-app.md
index 0979000b2a..ed710741de 100644
--- a/system_prompts/base/raw-app.md
+++ b/system_prompts/base/raw-app.md
@@ -253,6 +253,16 @@ frontend_api: [jobs:run, users:read]
 A file with a `sandbox` key states both settings: `sandbox: false` turns isolation off, and a missing `frontend_api` means no scopes. A file without the key leaves the deployed app's isolation and scopes as they are, so removing `sandbox: true` does not unsandbox it. `frontend_api` needs `sandbox: true`; the push fails otherwise.
 <!-- /cli-only -->
 
+## Reading data as the viewer
+
+A backend runnable reads data as the app's identity (the publisher in publisher mode), the same for every viewer. An inline runnable can instead read data as the person using the app: its data table connections that name no role hold that viewer's roles, and object storage permission rules (which paths may be listed, read, written) are checked against the viewer. Resources, variables and everything else stay the app's, and an anonymous viewer is refused. Turn it on only when what the runnable reads should differ per viewer: row-level security on a data table, or per-user paths in object storage. Both are Enterprise Edition features.
+<!-- chat-only -->
+Set it with `viewerDataAccess: true` on the backend runnable.
+<!-- /chat-only -->
+<!-- cli-only -->
+Set it with `viewer_data_access: true` in the runnable's `backend/<id>.yaml`.
+<!-- /cli-only -->
+
 ## Data Tables
 
 Data tables are PostgreSQL databases managed by Windmill. Backend runnables query them via the `wmill` client; the frontend never queries them directly. **When the app needs to store or persist data** (user data, settings, application state, records, logs), use a data table.
@@ -264,7 +274,7 @@ Data tables are PostgreSQL databases managed by Windmill. Backend runnables quer
 3. **No DDL inside runnables**: runnables only read and write rows (SELECT, INSERT, UPDATE, DELETE) on existing tables. Never CREATE, ALTER or DROP a table from a runnable.
 4. **Qualify table names**: an unqualified name means the `public` schema, so write every other table as `schema.table`, in table creation and queries alike. The app's `data` config sets the default datatable and schema its tables go in; use them consistently across runnables.
 5. **Pass the role**: when the app's `data.roles` gives a data table a role, every `wmill.datatable` call on it passes that role (`wmill.datatable('main', { role: 'analyst' })` in TypeScript, `wmill.datatable('main', role='analyst')` in Python). The role only reaches what it was granted, so a query outside it fails with `permission denied`.
-6. **Per-viewer rows**: a data table with no entry in `data.roles` is called with no role. Unless its default role is `admin`, the connection then holds the default role, if it has one, plus every role the person using the app is a tenant of, and is refused when that is none. Row-level security policies written as `pg_has_role(current_user, '<role>', 'member')` return that viewer's rows with nothing added to the runnable. A role pins the connection to that one role for every viewer, so where rows should differ per viewer give the data table no `data.roles` entry.
+6. **Per-viewer rows**: a data table with no entry in `data.roles` is called with no role. Unless its default role is `admin`, the connection then holds the default role, if it may use it, plus every role it is a tenant of, and is refused when that is none. Whose roles those are follows the runnable's identity (see "Reading data as the viewer"). Row-level security policies written as `pg_has_role(current_user, '<role>', 'member')` then return the right rows with nothing added to the code. A role pins the connection to that one role either way, so a data table whose rows should differ per viewer gets no `data.roles` entry. Data table roles are an Enterprise Edition feature, self-hosted only.
 <!-- cli-only -->
 
 `wmill datatable list` lists the workspace's data tables. Create or change tables with a migration in `sql_to_apply/` (see "SQL Migrations" above), then add them to `data.tables` in `raw_app.yaml`.