diff --git a/apps/dev-playground/server/index.ts b/apps/dev-playground/server/index.ts
index a4b6a2c6..26e1c97f 100644
--- a/apps/dev-playground/server/index.ts
+++ b/apps/dev-playground/server/index.ts
@@ -1,5 +1,12 @@
 import "reflect-metadata";
-import { analytics, createApp, files, genie, server } from "@databricks/appkit";
+import {
+  analytics,
+  createApp,
+  files,
+  genie,
+  policy,
+  server,
+} from "@databricks/appkit";
 import { WorkspaceClient } from "@databricks/sdk-experimental";
 import { lakebaseExamples } from "./lakebase-examples-plugin";
 import { reconnect } from "./reconnect-plugin";
@@ -25,7 +32,7 @@ createApp({
       spaces: { demo: process.env.DATABRICKS_GENIE_SPACE_ID ?? "placeholder" },
     }),
     lakebaseExamples(),
-    files(),
+    files({ volumes: { default: { policy: policy.allowAll() } } }),
   ],
   ...(process.env.APPKIT_E2E_TEST && { client: createMockClient() }),
 }).then((appkit) => {
diff --git a/docs/docs/api/appkit/Variable.policy.md b/docs/docs/api/appkit/Variable.policy.md
new file mode 100644
index 00000000..f1a00143
--- /dev/null
+++ b/docs/docs/api/appkit/Variable.policy.md
@@ -0,0 +1,119 @@
+# Variable: policy
+
+```ts
+const policy: {
+  all: FilePolicy;
+  allowAll: FilePolicy;
+  any: FilePolicy;
+  denyAll: FilePolicy;
+  not: FilePolicy;
+  publicRead: FilePolicy;
+  publicReadAndList: FilePolicy;
+};
+```
+
+Utility namespace with common policy combinators.
+
+## Type Declaration
+
+### all()
+
+```ts
+readonly all(...policies: FilePolicy[]): FilePolicy;
+```
+
+AND — all policies must allow. Short-circuits on first denial.
+
+#### Parameters
+
+| Parameter | Type |
+| ------ | ------ |
+| ...`policies` | `FilePolicy`[] |
+
+#### Returns
+
+`FilePolicy`
+
+### allowAll()
+
+```ts
+readonly allowAll(): FilePolicy;
+```
+
+Allow every action.
+
+#### Returns
+
+`FilePolicy`
+
+### any()
+
+```ts
+readonly any(...policies: FilePolicy[]): FilePolicy;
+```
+
+OR — at least one policy must allow. Short-circuits on first allow.
+
+#### Parameters
+
+| Parameter | Type |
+| ------ | ------ |
+| ...`policies` | `FilePolicy`[] |
+
+#### Returns
+
+`FilePolicy`
+
+### denyAll()
+
+```ts
+readonly denyAll(): FilePolicy;
+```
+
+Deny every action.
+
+#### Returns
+
+`FilePolicy`
+
+### not()
+
+```ts
+readonly not(p: FilePolicy): FilePolicy;
+```
+
+Negates a policy.
+
+#### Parameters
+
+| Parameter | Type |
+| ------ | ------ |
+| `p` | `FilePolicy` |
+
+#### Returns
+
+`FilePolicy`
+
+### publicRead()
+
+```ts
+readonly publicRead(): FilePolicy;
+```
+
+Allow all read actions (list, read, download, raw, exists, metadata, preview).
+
+#### Returns
+
+`FilePolicy`
+
+### publicReadAndList()
+
+```ts
+readonly publicReadAndList(): FilePolicy;
+```
+
+Alias for `publicRead()` — included for discoverability.
+
+#### Returns
+
+`FilePolicy`
diff --git a/docs/docs/api/appkit/index.md b/docs/docs/api/appkit/index.md
index 4ad80820..2c54db93 100644
--- a/docs/docs/api/appkit/index.md
+++ b/docs/docs/api/appkit/index.md
@@ -60,6 +60,7 @@ plugin architecture, and React integration.
 
 | Variable | Description |
 | ------ | ------ |
+| [policy](Variable.policy.md) | Utility namespace with common policy combinators. |
 | [sql](Variable.sql.md) | SQL helper namespace |
 
 ## Functions
diff --git a/docs/docs/api/appkit/typedoc-sidebar.ts b/docs/docs/api/appkit/typedoc-sidebar.ts
index 2f17b1d2..1a23d2c4 100644
--- a/docs/docs/api/appkit/typedoc-sidebar.ts
+++ b/docs/docs/api/appkit/typedoc-sidebar.ts
@@ -194,6 +194,11 @@ const typedocSidebar: SidebarsConfig = {
       type: "category",
       label: "Variables",
       items: [
+        {
+          type: "doc",
+          id: "api/appkit/Variable.policy",
+          label: "policy"
+        },
         {
           type: "doc",
           id: "api/appkit/Variable.sql",
diff --git a/docs/docs/plugins/files.md b/docs/docs/plugins/files.md
index ccc91265..b57f46e0 100644
--- a/docs/docs/plugins/files.md
+++ b/docs/docs/plugins/files.md
@@ -15,6 +15,7 @@ File operations against Databricks Unity Catalog Volumes. Supports listing, read
 - Automatic cache invalidation on write operations
 - Custom content type mappings
 - Per-user execution context (OBO)
+- **Access policies**: Per-volume policy functions that gate read and write operations
 
 ## Basic usage
 
@@ -75,6 +76,8 @@ interface IFilesConfig {
 }
 
 interface VolumeConfig {
+  /** Access policy for this volume. */
+  policy?: FilePolicy;
   /** Maximum upload size in bytes for this volume. Overrides plugin-level default. */
   maxUploadSize?: number;
   /** Map of file extensions to MIME types for this volume. Overrides plugin-level default. */
@@ -97,6 +100,115 @@ files({
 });
 ```
 
+### Permission model
+
+There are three layers of access control in the files plugin. Understanding how they interact is critical for securing your app:
+
+```
+┌─────────────────────────────────────────────────┐
+│  Unity Catalog grants                           │
+│  (WRITE_VOLUME on the SP — set at deploy time)  │
+├─────────────────────────────────────────────────┤
+│  Execution identity                             │
+│  HTTP routes → always service principal         │
+│  Programmatic → SP by default, asUser() for OBO │
+├─────────────────────────────────────────────────┤
+│  File policies                                  │
+│  Per-volume (action, resource, user) → boolean  │
+│  Only app-level gate for HTTP routes            │
+└─────────────────────────────────────────────────┘
+```
+
+- **UC grants** control what the service principal can do at the Databricks level. These are set at deploy time via `app.yaml` resource bindings. The SP needs `WRITE_VOLUME` — the plugin declares this via resource requirements.
+- **Execution identity** determines whose credentials are used for the actual API call. HTTP routes always use the SP. The programmatic API uses SP by default but supports `asUser(req)` for OBO.
+- **File policies** are application-level checks evaluated **before** the API call. They receive the requesting user's identity (from the `x-forwarded-user` header) and decide allow/deny. This is the only gate that distinguishes between users on HTTP routes.
+
+:::warning
+
+Since HTTP routes always execute as the service principal, removing a user's UC `WRITE_VOLUME` grant has **no effect** on HTTP access — the SP's grant is what's used. Policies are how you restrict what individual users can do through your app.
+
+:::
+
+#### Access policies
+
+Attach a policy to a volume to control which actions are allowed:
+
+```ts
+import { files, policy } from "@databricks/appkit";
+
+files({
+  volumes: {
+    uploads: { policy: policy.publicRead() },
+  },
+});
+```
+
+#### Actions
+
+Policies receive an action string. The full list, split by category:
+
+| Category | Actions |
+|----------|---------|
+| Read | `list`, `read`, `download`, `raw`, `exists`, `metadata`, `preview` |
+| Write | `upload`, `mkdir`, `delete` |
+
+#### Built-in policies
+
+| Helper | Allows | Denies |
+|--------|--------|--------|
+| `policy.publicRead()` | all read actions | all write actions |
+| `policy.allowAll()` | everything | nothing |
+| `policy.denyAll()` | nothing | everything |
+
+#### Composing policies
+
+Combine built-in and custom policies with three combinators:
+
+- **`policy.all(a, b)`** — AND: all policies must allow. Short-circuits on first denial.
+- **`policy.any(a, b)`** — OR: at least one policy must allow. Short-circuits on first allow.
+- **`policy.not(p)`** — Inverts a policy. For example, `not(publicRead())` yields a write-only policy (useful for ingestion/drop-box volumes).
+
+```ts
+// Read-only for regular users, full access for the service principal
+files({
+  volumes: {
+    shared: {
+      policy: policy.any(
+        (_action, _resource, user) => !!user.isServicePrincipal,
+        policy.publicRead(),
+      ),
+    },
+  },
+});
+```
+
+#### Custom policies
+
+`FilePolicy` is a function `(action, resource, user) → boolean | Promise<boolean>`, so you can inline arbitrary logic:
+
+```ts
+import { type FilePolicy, WRITE_ACTIONS } from "@databricks/appkit";
+
+const ADMIN_IDS = ["admin-sp-id", "lead-user-id"];
+
+const adminOnly: FilePolicy = (action, _resource, user) => {
+  if (WRITE_ACTIONS.has(action)) {
+    return ADMIN_IDS.includes(user.id);
+  }
+  return true; // reads allowed for everyone
+};
+
+files({
+  volumes: { reports: { policy: adminOnly } },
+});
+```
+
+#### Enforcement
+
+- **HTTP routes**: Policy checked before every operation. Denied → `403` JSON response with `"Action denied by volume policy"`.
+- **Programmatic API**: Policy checked on both `appkit.files("vol").list()` (SP identity, `isServicePrincipal: true`) and `appkit.files("vol").asUser(req).list()` (user identity). Denied → throws `PolicyDeniedError`.
+- **No policy configured**: Defaults to `publicRead()` — read actions are allowed, write actions are denied. A startup warning is logged encouraging you to set an explicit policy.
+
 ### Custom content types
 
 Override or extend the built-in extension → MIME map:
@@ -236,7 +348,36 @@ interface FilePreview extends FileMetadata {
   isImage: boolean;
 }
 
+type FileAction =
+  | "list" | "read" | "download" | "raw"
+  | "exists" | "metadata" | "preview"
+  | "upload" | "mkdir" | "delete";
+
+interface FileResource {
+  /** Relative path within the volume. */
+  path: string;
+  /** The volume key (e.g. `"uploads"`). */
+  volume: string;
+  /** Content length in bytes — only present for uploads. */
+  size?: number;
+}
+
+interface FilePolicyUser {
+  /** User ID from the `x-forwarded-user` header. */
+  id: string;
+  /** `true` when the caller is the service principal (direct SDK call, not `asUser`). */
+  isServicePrincipal?: boolean;
+}
+
+type FilePolicy = (
+  action: FileAction,
+  resource: FileResource,
+  user: FilePolicyUser,
+) => boolean | Promise<boolean>;
+
 interface VolumeConfig {
+  /** Access policy for this volume. */
+  policy?: FilePolicy;
   /** Maximum upload size in bytes for this volume. */
   maxUploadSize?: number;
   /** Map of file extensions to MIME types for this volume. */
@@ -297,6 +438,7 @@ All errors return JSON:
 | Status | Description                                                    |
 | ------ | -------------------------------------------------------------- |
 | 400    | Missing or invalid `path` parameter                            |
+| 403    | Policy denied "`{action}`" on volume "`{volumeKey}`"          |
 | 404    | Unknown volume key                                             |
 | 413    | Upload exceeds `maxUploadSize`                                 |
 | 500    | Operation failed (SDK, network, upstream, or unhandled error)  |
diff --git a/docs/static/appkit-ui/styles.gen.css b/docs/static/appkit-ui/styles.gen.css
index 419e0500..eb938427 100644
--- a/docs/static/appkit-ui/styles.gen.css
+++ b/docs/static/appkit-ui/styles.gen.css
@@ -5578,7 +5578,10 @@
   }
   ::selection {
     background-color: var(--primary);
-    opacity: 0.2;
+    @supports (color: color-mix(in lab, red, red)) {
+      background-color: color-mix(in oklch, var(--primary) 30%, transparent);
+    }
+    color: var(--primary-foreground);
   }
   @media (prefers-reduced-motion: reduce) {
     *,
diff --git a/packages/appkit/src/context/execution-context.ts b/packages/appkit/src/context/execution-context.ts
index d707f52d..dbc4788e 100644
--- a/packages/appkit/src/context/execution-context.ts
+++ b/packages/appkit/src/context/execution-context.ts
@@ -81,11 +81,3 @@ export function getWarehouseId(): Promise<string> {
 export function getWorkspaceId(): Promise<string> {
   return getExecutionContext().workspaceId;
 }
-
-/**
- * Check if currently running in a user context.
- */
-export function isInUserContext(): boolean {
-  const ctx = executionContextStorage.getStore();
-  return ctx !== undefined;
-}
diff --git a/packages/appkit/src/context/index.ts b/packages/appkit/src/context/index.ts
index 6a5a738b..d306d359 100644
--- a/packages/appkit/src/context/index.ts
+++ b/packages/appkit/src/context/index.ts
@@ -4,7 +4,6 @@ export {
   getWarehouseId,
   getWorkspaceClient,
   getWorkspaceId,
-  isInUserContext,
   runInUserContext,
 } from "./execution-context";
 export { ServiceContext } from "./service-context";
diff --git a/packages/appkit/src/context/user-context.ts b/packages/appkit/src/context/user-context.ts
index 20746c91..12548103 100644
--- a/packages/appkit/src/context/user-context.ts
+++ b/packages/appkit/src/context/user-context.ts
@@ -20,13 +20,13 @@ export interface UserContext {
 }
 
 /**
- * Execution context can be either service or user context.
- */
-export type ExecutionContext = ServiceContextState | UserContext;
-
-/**
- * Check if an execution context is a user context.
+ * Type guard to check if a context is a UserContext.
  */
 export function isUserContext(ctx: ExecutionContext): ctx is UserContext {
   return "isUserContext" in ctx && ctx.isUserContext === true;
 }
+
+/**
+ * Execution context can be either service or user context.
+ */
+export type ExecutionContext = ServiceContextState | UserContext;
diff --git a/packages/appkit/src/index.ts b/packages/appkit/src/index.ts
index 8db7f1d7..6db063fa 100644
--- a/packages/appkit/src/index.ts
+++ b/packages/appkit/src/index.ts
@@ -49,6 +49,7 @@ export {
 // Plugin authoring
 export { Plugin, type ToPlugin, toPlugin } from "./plugin";
 export { analytics, files, genie, lakebase, server } from "./plugins";
+export { policy } from "./plugins/files";
 // Registry types and utilities for plugin manifests
 export type {
   ConfigSchema,
diff --git a/packages/appkit/src/plugins/files/index.ts b/packages/appkit/src/plugins/files/index.ts
index 386c47ef..64a38714 100644
--- a/packages/appkit/src/plugins/files/index.ts
+++ b/packages/appkit/src/plugins/files/index.ts
@@ -1,3 +1,4 @@
 export * from "./defaults";
 export * from "./plugin";
+export * from "./policy";
 export * from "./types";
diff --git a/packages/appkit/src/plugins/files/plugin.ts b/packages/appkit/src/plugins/files/plugin.ts
index e10f5d42..96eae020 100644
--- a/packages/appkit/src/plugins/files/plugin.ts
+++ b/packages/appkit/src/plugins/files/plugin.ts
@@ -8,7 +8,7 @@ import {
   isSafeInlineContentType,
   validateCustomContentTypes,
 } from "../../connectors/files";
-import { getWorkspaceClient, isInUserContext } from "../../context";
+import { getCurrentUserId, getWorkspaceClient } from "../../context";
 import { AuthenticationError } from "../../errors";
 import { createLogger } from "../../logging/logger";
 import { Plugin, toPlugin } from "../../plugin";
@@ -22,6 +22,13 @@ import {
 } from "./defaults";
 import { parentDirectory, sanitizeFilename } from "./helpers";
 import manifest from "./manifest.json";
+import {
+  type FileAction,
+  type FilePolicyUser,
+  type FileResource,
+  PolicyDeniedError,
+  policy,
+} from "./policy";
 import type {
   DownloadResponse,
   FilesExport,
@@ -91,29 +98,107 @@ export class FilesPlugin extends Plugin {
     }));
   }
 
+  /** Whether a volume has a policy attached. */
+  private _hasPolicy(volumeKey: string): boolean {
+    return typeof this.volumeConfigs[volumeKey]?.policy === "function";
+  }
+
   /**
-   * Warns when a method is called without a user context (i.e. as service principal).
-   * OBO access via `asUser(req)` is strongly recommended.
+   * Extract user identity from the request.
+   * Falls back to `getCurrentUserId()` in development mode.
    */
-  private warnIfNoUserContext(volumeKey: string, method: string): void {
-    if (!isInUserContext()) {
+  private _extractUser(req: import("express").Request): FilePolicyUser {
+    const userId = req.header("x-forwarded-user");
+    if (userId) return { id: userId };
+    if (process.env.NODE_ENV === "development") {
       logger.warn(
-        `app.files("${volumeKey}").${method}() called without user context (service principal). ` +
-          `Please use OBO instead: app.files("${volumeKey}").asUser(req).${method}()`,
+        "No x-forwarded-user header — falling back to service principal identity for policy checks. " +
+          "Ensure your proxy forwards user headers to test per-user policies.",
       );
+      return { id: getCurrentUserId() };
     }
+    throw AuthenticationError.missingToken(
+      "Missing x-forwarded-user header. Cannot resolve user ID.",
+    );
   }
 
   /**
-   * Throws when a method is called without a user context (i.e. as service principal).
-   * OBO access via `asUser(req)` is enforced for now.
+   * Check the policy for a volume. No-op if no policy is configured.
+   * Throws `PolicyDeniedError` if denied.
    */
-  private throwIfNoUserContext(volumeKey: string, method: string): void {
-    if (!isInUserContext()) {
-      throw new Error(
-        `app.files("${volumeKey}").${method}() called without user context (service principal). Use OBO instead: app.files("${volumeKey}").asUser(req).${method}()`,
-      );
+  private async _checkPolicy(
+    volumeKey: string,
+    action: FileAction,
+    path: string,
+    user: FilePolicyUser,
+    resourceOverrides?: Partial<FileResource>,
+  ): Promise<void> {
+    const policyFn = this.volumeConfigs[volumeKey]?.policy;
+    if (typeof policyFn !== "function") return;
+
+    const resource: FileResource = {
+      path,
+      volume: volumeKey,
+      ...resourceOverrides,
+    };
+    const allowed = await policyFn(action, resource, user);
+    if (!allowed) {
+      try {
+        logger.warn(
+          'Policy denied "%s" on volume "%s" for user "%s"',
+          action,
+          volumeKey,
+          user.id,
+        );
+      } catch {
+        // user.id may not be resolvable in all contexts (e.g. lazy SP getter)
+        logger.warn('Policy denied "%s" on volume "%s"', action, volumeKey);
+      }
+      throw new PolicyDeniedError(action, volumeKey);
+    }
+  }
+
+  /**
+   * HTTP-level wrapper around `_checkPolicy`.
+   * Extracts user (401 on failure), runs policy (403 on denial).
+   * Returns `true` if the request may proceed, `false` if a response was sent.
+   */
+  private async _enforcePolicy(
+    req: import("express").Request,
+    res: import("express").Response,
+    volumeKey: string,
+    action: FileAction,
+    resourceOverrides?: Partial<FileResource>,
+  ): Promise<boolean> {
+    if (!this._hasPolicy(volumeKey)) return true;
+
+    let user: FilePolicyUser;
+    try {
+      user = this._extractUser(req);
+    } catch (error) {
+      if (error instanceof AuthenticationError) {
+        res.status(401).json({ error: error.message, plugin: this.name });
+        return false;
+      }
+      throw error;
+    }
+
+    const path =
+      (req.query.path as string | undefined) ??
+      (typeof req.body?.path === "string" ? req.body.path : undefined) ??
+      "/";
+
+    try {
+      await this._checkPolicy(volumeKey, action, path, user, resourceOverrides);
+    } catch (error) {
+      if (error instanceof PolicyDeniedError) {
+        res.status(403).json({ error: error.message, plugin: this.name });
+        return false;
+      }
+      throw error;
     }
+
+    return true;
   }
 
   constructor(config: IFilesConfig) {
@@ -137,6 +222,7 @@ export class FilesPlugin extends Plugin {
         maxUploadSize: volumeCfg.maxUploadSize ?? config.maxUploadSize,
         customContentTypes:
           volumeCfg.customContentTypes ?? config.customContentTypes,
+        policy: volumeCfg.policy ?? policy.publicRead(),
       };
       this.volumeConfigs[key] = mergedConfig;
 
@@ -147,60 +233,48 @@ export class FilesPlugin extends Plugin {
         customContentTypes: mergedConfig.customContentTypes,
       });
     }
+
+    // Warn at startup for volumes without an explicit policy
+    for (const key of this.volumeKeys) {
+      if (!volumes[key].policy) {
+        logger.warn(
+          'Volume "%s" has no explicit policy — defaulting to publicRead(). ' +
+            "Set a policy in files({ volumes: { %s: { policy: ... } } }) to silence this warning.",
+          key,
+          key,
+        );
+      }
+    }
   }
 
   /**
    * Creates a VolumeAPI for a specific volume key.
-   * Each method warns if called outside a user context (service principal).
+   * All operations execute as the service principal.
    */
   protected createVolumeAPI(volumeKey: string): VolumeAPI {
     const connector = this.volumeConnectors[volumeKey];
     return {
-      list: (directoryPath?: string) => {
-        this.throwIfNoUserContext(volumeKey, `list`);
-        return connector.list(getWorkspaceClient(), directoryPath);
-      },
-      read: (filePath: string, options?: { maxSize?: number }) => {
-        this.throwIfNoUserContext(volumeKey, `read`);
-        return connector.read(getWorkspaceClient(), filePath, options);
-      },
-      download: (filePath: string): Promise<DownloadResponse> => {
-        this.throwIfNoUserContext(volumeKey, `download`);
-        return connector.download(getWorkspaceClient(), filePath);
-      },
-      exists: (filePath: string) => {
-        this.throwIfNoUserContext(volumeKey, `exists`);
-        return connector.exists(getWorkspaceClient(), filePath);
-      },
-      metadata: (filePath: string) => {
-        this.throwIfNoUserContext(volumeKey, `metadata`);
-        return connector.metadata(getWorkspaceClient(), filePath);
-      },
+      list: (directoryPath?: string) =>
+        connector.list(getWorkspaceClient(), directoryPath),
+      read: (filePath: string, options?: { maxSize?: number }) =>
+        connector.read(getWorkspaceClient(), filePath, options),
+      download: (filePath: string): Promise<DownloadResponse> =>
+        connector.download(getWorkspaceClient(), filePath),
+      exists: (filePath: string) =>
+        connector.exists(getWorkspaceClient(), filePath),
+      metadata: (filePath: string) =>
+        connector.metadata(getWorkspaceClient(), filePath),
       upload: (
         filePath: string,
         contents: ReadableStream | Buffer | string,
         options?: { overwrite?: boolean },
-      ) => {
-        this.throwIfNoUserContext(volumeKey, `upload`);
-        return connector.upload(
-          getWorkspaceClient(),
-          filePath,
-          contents,
-          options,
-        );
-      },
-      createDirectory: (directoryPath: string) => {
-        this.throwIfNoUserContext(volumeKey, `createDirectory`);
-        return connector.createDirectory(getWorkspaceClient(), directoryPath);
-      },
-      delete: (filePath: string) => {
-        this.throwIfNoUserContext(volumeKey, `delete`);
-        return connector.delete(getWorkspaceClient(), filePath);
-      },
-      preview: (filePath: string) => {
-        this.throwIfNoUserContext(volumeKey, `preview`);
-        return connector.preview(getWorkspaceClient(), filePath);
-      },
+      ) => connector.upload(getWorkspaceClient(), filePath, contents, options),
+      createDirectory: (directoryPath: string) =>
+        connector.createDirectory(getWorkspaceClient(), directoryPath),
+      delete: (filePath: string) =>
+        connector.delete(getWorkspaceClient(), filePath),
+      preview: (filePath: string) =>
+        connector.preview(getWorkspaceClient(), filePath),
     };
   }
 
@@ -380,7 +454,6 @@ export class FilesPlugin extends Plugin {
   private _invalidateListCache(
     volumeKey: string,
     parentPath: string,
-    userId: string,
     connector: FilesConnector,
   ): void {
     const parent = parentDirectory(parentPath);
@@ -389,16 +462,63 @@ export class FilesPlugin extends Plugin {
       : "__root__";
     const listKey = this.cache.generateKey(
       [`files:${volumeKey}:list`, cachePathSegment],
-      userId,
+      getCurrentUserId(),
     );
     this.cache.delete(listKey);
   }
 
+  /**
+   * Like `execute()`, but re-throws Databricks API client errors (4xx)
+   * so route handlers can return the appropriate HTTP status via `_handleApiError`.
+   *
+   * Server errors and infrastructure failures are still swallowed (returns `undefined`).
+   */
+  private async _executeOrThrow<T>(
+    fn: (signal?: AbortSignal) => Promise<T>,
+    options: PluginExecutionSettings,
+    userKey?: string,
+  ): Promise<T | undefined> {
+    let capturedClientError: unknown;
+
+    const result = await this.execute<T>(
+      async (signal) => {
+        try {
+          return await fn(signal);
+        } catch (error) {
+          if (
+            error instanceof ApiError &&
+            error.statusCode !== undefined &&
+            error.statusCode >= 400 &&
+            error.statusCode < 500
+          ) {
+            capturedClientError = error;
+          }
+          throw error;
+        }
+      },
+      options,
+      userKey,
+    );
+
+    if (result === undefined && capturedClientError) {
+      throw capturedClientError;
+    }
+
+    return result;
+  }
+
   private _handleApiError(
     res: express.Response,
     error: unknown,
     fallbackMessage: string,
   ): void {
+    if (error instanceof PolicyDeniedError) {
+      res.status(403).json({
+        error: error.message,
+        plugin: this.name,
+      });
+      return;
+    }
     if (error instanceof AuthenticationError) {
       res.status(401).json({
         error: error.message,
@@ -430,15 +550,13 @@ export class FilesPlugin extends Plugin {
     connector: FilesConnector,
     volumeKey: string,
   ): Promise<void> {
+    if (!(await this._enforcePolicy(req, res, volumeKey, "list"))) return;
+
     const path = req.query.path as string | undefined;
 
     try {
-      const userPlugin = this.asUser(req);
-      const result = await userPlugin.execute(
-        async () => {
-          this.warnIfNoUserContext(volumeKey, `list`);
-          return connector.list(getWorkspaceClient(), path);
-        },
+      const result = await this.execute(
+        async () => connector.list(getWorkspaceClient(), path),
         this._readSettings([
           `files:${volumeKey}:list`,
           path ? connector.resolvePath(path) : "__root__",
@@ -461,6 +579,8 @@ export class FilesPlugin extends Plugin {
     connector: FilesConnector,
     volumeKey: string,
   ): Promise<void> {
+    if (!(await this._enforcePolicy(req, res, volumeKey, "read"))) return;
+
     const path = req.query.path as string;
     const valid = this._isValidPath(path);
     if (valid !== true) {
@@ -469,12 +589,8 @@ export class FilesPlugin extends Plugin {
     }
 
     try {
-      const userPlugin = this.asUser(req);
-      const result = await userPlugin.execute(
-        async () => {
-          this.warnIfNoUserContext(volumeKey, `read`);
-          return connector.read(getWorkspaceClient(), path);
-        },
+      const result = await this.execute(
+        async () => connector.read(getWorkspaceClient(), path),
         this._readSettings([
           `files:${volumeKey}:read`,
           connector.resolvePath(path),
@@ -525,6 +641,8 @@ export class FilesPlugin extends Plugin {
     volumeKey: string,
     opts: { mode: "download" | "raw" },
   ): Promise<void> {
+    if (!(await this._enforcePolicy(req, res, volumeKey, opts.mode))) return;
+
     const path = req.query.path as string;
     const valid = this._isValidPath(path);
     if (valid !== true) {
@@ -536,14 +654,13 @@ export class FilesPlugin extends Plugin {
     const volumeCfg = this.volumeConfigs[volumeKey];
 
     try {
-      const userPlugin = this.asUser(req);
       const settings: PluginExecutionSettings = {
         default: FILES_DOWNLOAD_DEFAULTS,
       };
-      const response = await userPlugin.execute(async () => {
-        this.warnIfNoUserContext(volumeKey, `download`);
-        return connector.download(getWorkspaceClient(), path);
-      }, settings);
+      const response = await this.execute(
+        async () => connector.download(getWorkspaceClient(), path),
+        settings,
+      );
 
       if (response === undefined) {
         res.status(500).json({ error: `${label} failed`, plugin: this.name });
@@ -604,6 +721,8 @@ export class FilesPlugin extends Plugin {
     connector: FilesConnector,
     volumeKey: string,
   ): Promise<void> {
+    if (!(await this._enforcePolicy(req, res, volumeKey, "exists"))) return;
+
     const path = req.query.path as string;
     const valid = this._isValidPath(path);
     if (valid !== true) {
@@ -612,12 +731,8 @@ export class FilesPlugin extends Plugin {
     }
 
     try {
-      const userPlugin = this.asUser(req);
-      const result = await userPlugin.execute(
-        async () => {
-          this.warnIfNoUserContext(volumeKey, `exists`);
-          return connector.exists(getWorkspaceClient(), path);
-        },
+      const result = await this.execute(
+        async () => connector.exists(getWorkspaceClient(), path),
         this._readSettings([
           `files:${volumeKey}:exists`,
           connector.resolvePath(path),
@@ -642,6 +757,8 @@ export class FilesPlugin extends Plugin {
     connector: FilesConnector,
     volumeKey: string,
   ): Promise<void> {
+    if (!(await this._enforcePolicy(req, res, volumeKey, "metadata"))) return;
+
     const path = req.query.path as string;
     const valid = this._isValidPath(path);
     if (valid !== true) {
@@ -650,12 +767,8 @@ export class FilesPlugin extends Plugin {
     }
 
     try {
-      const userPlugin = this.asUser(req);
-      const result = await userPlugin.execute(
-        async () => {
-          this.warnIfNoUserContext(volumeKey, `metadata`);
-          return connector.metadata(getWorkspaceClient(), path);
-        },
+      const result = await this.execute(
+        async () => connector.metadata(getWorkspaceClient(), path),
         this._readSettings([
           `files:${volumeKey}:metadata`,
           connector.resolvePath(path),
@@ -680,6 +793,8 @@ export class FilesPlugin extends Plugin {
     connector: FilesConnector,
     volumeKey: string,
   ): Promise<void> {
+    if (!(await this._enforcePolicy(req, res, volumeKey, "preview"))) return;
+
     const path = req.query.path as string;
     const valid = this._isValidPath(path);
     if (valid !== true) {
@@ -688,12 +803,8 @@ export class FilesPlugin extends Plugin {
     }
 
     try {
-      const userPlugin = this.asUser(req);
-      const result = await userPlugin.execute(
-        async () => {
-          this.warnIfNoUserContext(volumeKey, `preview`);
-          return connector.preview(getWorkspaceClient(), path);
-        },
+      const result = await this.execute(
+        async () => connector.preview(getWorkspaceClient(), path),
         this._readSettings([
           `files:${volumeKey}:preview`,
           connector.resolvePath(path),
@@ -742,6 +853,13 @@ export class FilesPlugin extends Plugin {
       return;
     }
 
+    if (
+      !(await this._enforcePolicy(req, res, volumeKey, "upload", {
+        size: contentLength,
+      }))
+    )
+      return;
+
     logger.debug(req, "Upload started: volume=%s path=%s", volumeKey, path);
 
     try {
@@ -772,24 +890,17 @@ export class FilesPlugin extends Plugin {
         path,
         contentLength ?? 0,
       );
-      const userPlugin = this.asUser(req);
       const settings: PluginExecutionSettings = {
         default: FILES_WRITE_DEFAULTS,
       };
       const result = await this.trackWrite(() =>
-        userPlugin.execute(async () => {
-          this.warnIfNoUserContext(volumeKey, `upload`);
+        this.execute(async () => {
           await connector.upload(getWorkspaceClient(), path, webStream);
           return { success: true as const };
         }, settings),
       );
 
-      this._invalidateListCache(
-        volumeKey,
-        path,
-        this.resolveUserId(req),
-        connector,
-      );
+      this._invalidateListCache(volumeKey, path, connector);
 
       if (result === undefined) {
         logger.error(
@@ -823,6 +934,8 @@ export class FilesPlugin extends Plugin {
     connector: FilesConnector,
     volumeKey: string,
   ): Promise<void> {
+    if (!(await this._enforcePolicy(req, res, volumeKey, "mkdir"))) return;
+
     const dirPath =
       typeof req.body?.path === "string" ? req.body.path : undefined;
     const valid = this._isValidPath(dirPath);
@@ -832,24 +945,17 @@ export class FilesPlugin extends Plugin {
     }
 
     try {
-      const userPlugin = this.asUser(req);
       const settings: PluginExecutionSettings = {
         default: FILES_WRITE_DEFAULTS,
       };
       const result = await this.trackWrite(() =>
-        userPlugin.execute(async () => {
-          this.warnIfNoUserContext(volumeKey, `createDirectory`);
+        this.execute(async () => {
           await connector.createDirectory(getWorkspaceClient(), dirPath);
           return { success: true as const };
         }, settings),
       );
 
-      this._invalidateListCache(
-        volumeKey,
-        dirPath,
-        this.resolveUserId(req),
-        connector,
-      );
+      this._invalidateListCache(volumeKey, dirPath, connector);
 
       if (result === undefined) {
         res
@@ -870,6 +976,8 @@ export class FilesPlugin extends Plugin {
     connector: FilesConnector,
     volumeKey: string,
   ): Promise<void> {
+    if (!(await this._enforcePolicy(req, res, volumeKey, "delete"))) return;
+
     const rawPath = req.query.path as string | undefined;
     const valid = this._isValidPath(rawPath);
     if (valid !== true) {
@@ -879,24 +987,17 @@ export class FilesPlugin extends Plugin {
     const path = rawPath as string;
 
     try {
-      const userPlugin = this.asUser(req);
       const settings: PluginExecutionSettings = {
         default: FILES_WRITE_DEFAULTS,
       };
       const result = await this.trackWrite(() =>
-        userPlugin.execute(async () => {
-          this.warnIfNoUserContext(volumeKey, `delete`);
+        this.execute(async () => {
           await connector.delete(getWorkspaceClient(), path);
           return { success: true as const };
         }, settings),
       );
 
-      this._invalidateListCache(
-        volumeKey,
-        path,
-        this.resolveUserId(req),
-        connector,
-      );
+      this._invalidateListCache(volumeKey, path, connector);
 
       if (result === undefined) {
         res.status(500).json({ error: "Delete failed", plugin: this.name });
@@ -909,6 +1010,70 @@ export class FilesPlugin extends Plugin {
     }
   }
 
+  /**
+   * Creates a VolumeAPI that enforces the volume's policy and then delegates
+   * to the connector as the service principal.
+   */
+  private _createPolicyWrappedAPI(
+    volumeKey: string,
+    user: FilePolicyUser,
+  ): VolumeAPI {
+    const connector = this.volumeConnectors[volumeKey];
+    const check = (
+      action: FileAction,
+      path: string,
+      overrides?: Partial<FileResource>,
+    ) => this._checkPolicy(volumeKey, action, path, user, overrides);
+
+    return {
+      list: async (directoryPath?: string) => {
+        await check("list", directoryPath ?? "/");
+        return connector.list(getWorkspaceClient(), directoryPath);
+      },
+      read: async (filePath: string, options?: { maxSize?: number }) => {
+        await check("read", filePath);
+        return connector.read(getWorkspaceClient(), filePath, options);
+      },
+      download: async (filePath: string) => {
+        await check("download", filePath);
+        return connector.download(getWorkspaceClient(), filePath);
+      },
+      exists: async (filePath: string) => {
+        await check("exists", filePath);
+        return connector.exists(getWorkspaceClient(), filePath);
+      },
+      metadata: async (filePath: string) => {
+        await check("metadata", filePath);
+        return connector.metadata(getWorkspaceClient(), filePath);
+      },
+      upload: async (
+        filePath: string,
+        contents: ReadableStream | Buffer | string,
+        options?: { overwrite?: boolean },
+      ) => {
+        await check("upload", filePath);
+        return connector.upload(
+          getWorkspaceClient(),
+          filePath,
+          contents,
+          options,
+        );
+      },
+      createDirectory: async (directoryPath: string) => {
+        await check("mkdir", directoryPath);
+        return connector.createDirectory(getWorkspaceClient(), directoryPath);
+      },
+      delete: async (filePath: string) => {
+        await check("delete", filePath);
+        return connector.delete(getWorkspaceClient(), filePath);
+      },
+      preview: async (filePath: string) => {
+        await check("preview", filePath);
+        return connector.preview(getWorkspaceClient(), filePath);
+      },
+    };
+  }
+
   private inflightWrites = 0;
 
   private trackWrite<T>(fn: () => Promise<T>): Promise<T> {
@@ -941,13 +1106,16 @@ export class FilesPlugin extends Plugin {
    * Returns the programmatic API for the Files plugin.
    * Callable with a volume key to get a volume-scoped handle.
    *
+   * All operations execute as the service principal.
+   * Use policies to control per-user access.
+   *
    * @example
    * ```ts
-   * // OBO access (recommended)
-   * appKit.files("uploads").asUser(req).list()
-   *
-   * // Service principal access (logs a warning)
+   * // Service principal access
    * appKit.files("uploads").list()
+   *
+   * // With policy: pass user identity for access control
+   * appKit.files("uploads").asUser(req).list()
    * ```
    */
   exports(): FilesExport {
@@ -958,14 +1126,21 @@ export class FilesPlugin extends Plugin {
         );
       }
 
-      // Service principal API — each method logs a warning recommending OBO
-      const spApi = this.createVolumeAPI(volumeKey);
+      // Lazy user resolution: getCurrentUserId() is called when a method
+      // is invoked (policy check), not when exports() is called.
+      const spUser: FilePolicyUser = {
+        get id() {
+          return getCurrentUserId();
+        },
+        isServicePrincipal: true,
+      };
+      const spApi = this._createPolicyWrappedAPI(volumeKey, spUser);
 
       return {
         ...spApi,
         asUser: (req: import("express").Request) => {
-          const userPlugin = this.asUser(req) as FilesPlugin;
-          return userPlugin.createVolumeAPI(volumeKey);
+          const user = this._extractUser(req);
+          return this._createPolicyWrappedAPI(volumeKey, user);
         },
       };
     };
diff --git a/packages/appkit/src/plugins/files/policy.ts b/packages/appkit/src/plugins/files/policy.ts
new file mode 100644
index 00000000..fd4f3143
--- /dev/null
+++ b/packages/appkit/src/plugins/files/policy.ts
@@ -0,0 +1,152 @@
+/**
+ * Per-volume file access policies.
+ *
+ * A `FilePolicy` is a function that decides whether a given action on a
+ * resource is allowed for a specific user. When a policy is attached to a
+ * volume, the policy controls whether the action is allowed for the requesting user.
+ */
+
+// ---------------------------------------------------------------------------
+// Actions
+// ---------------------------------------------------------------------------
+
+/** Every action the files plugin can perform. */
+export type FileAction =
+  | "list"
+  | "read"
+  | "download"
+  | "raw"
+  | "exists"
+  | "metadata"
+  | "preview"
+  | "upload"
+  | "mkdir"
+  | "delete";
+
+/** Actions that only read data. */
+export const READ_ACTIONS: ReadonlySet<FileAction> = new Set<FileAction>([
+  "list",
+  "read",
+  "download",
+  "raw",
+  "exists",
+  "metadata",
+  "preview",
+]);
+
+/** Actions that mutate data. */
+export const WRITE_ACTIONS: ReadonlySet<FileAction> = new Set<FileAction>([
+  "upload",
+  "mkdir",
+  "delete",
+]);
+
+// ---------------------------------------------------------------------------
+// Resource & User
+// ---------------------------------------------------------------------------
+
+/** Describes the file or directory being acted upon. */
+export interface FileResource {
+  /** Relative path within the volume. */
+  path: string;
+  /** The volume key (e.g. `"uploads"`). */
+  volume: string;
+  /** Content length in bytes — only present for uploads. */
+  size?: number;
+}
+
+/** Minimal user identity passed to the policy function. */
+export interface FilePolicyUser {
+  id: string;
+  /** `true` when the caller is the service principal (direct SDK call, not `asUser`). */
+  isServicePrincipal?: boolean;
+}
+
+// ---------------------------------------------------------------------------
+// Policy function type
+// ---------------------------------------------------------------------------
+
+/**
+ * A policy function that decides whether `user` may perform `action` on
+ * `resource`. Return `true` to allow, `false` to deny.
+ */
+export type FilePolicy = (
+  action: FileAction,
+  resource: FileResource,
+  user: FilePolicyUser,
+) => boolean | Promise<boolean>;
+
+// ---------------------------------------------------------------------------
+// PolicyDeniedError
+// ---------------------------------------------------------------------------
+
+/**
+ * Thrown when a policy denies an action.
+ */
+export class PolicyDeniedError extends Error {
+  readonly action: FileAction;
+  readonly volumeKey: string;
+
+  constructor(action: FileAction, volumeKey: string) {
+    super(`Policy denied "${action}" on volume "${volumeKey}"`);
+    this.name = "PolicyDeniedError";
+    this.action = action;
+    this.volumeKey = volumeKey;
+  }
+}
+
+// ---------------------------------------------------------------------------
+// Combinators
+// ---------------------------------------------------------------------------
+
+/** Utility namespace with common policy combinators. */
+export const policy = {
+  /**
+   * AND — all policies must allow. Short-circuits on first denial.
+   */
+  all(...policies: FilePolicy[]): FilePolicy {
+    return async (action, resource, user) => {
+      for (const p of policies) {
+        if (!(await p(action, resource, user))) return false;
+      }
+      return true;
+    };
+  },
+
+  /**
+   * OR — at least one policy must allow. Short-circuits on first allow.
+   */
+  any(...policies: FilePolicy[]): FilePolicy {
+    return async (action, resource, user) => {
+      for (const p of policies) {
+        if (await p(action, resource, user)) return true;
+      }
+      return false;
+    };
+  },
+
+  /** Negates a policy. */
+  not(p: FilePolicy): FilePolicy {
+    return async (action, resource, user) => !(await p(action, resource, user));
+  },
+
+  /** Allow all read actions (list, read, download, raw, exists, metadata, preview). */
+  publicRead(): FilePolicy {
+    return (action) => READ_ACTIONS.has(action);
+  },
+
+  /** Alias for `publicRead()` — included for discoverability. */
+  publicReadAndList(): FilePolicy {
+    return policy.publicRead();
+  },
+
+  /** Deny every action. */
+  denyAll(): FilePolicy {
+    return () => false;
+  },
+
+  /** Allow every action. */
+  allowAll(): FilePolicy {
+    return () => true;
+  },
+} as const;
diff --git a/packages/appkit/src/plugins/files/tests/plugin.integration.test.ts b/packages/appkit/src/plugins/files/tests/plugin.integration.test.ts
index e0117cdc..c979ae0e 100644
--- a/packages/appkit/src/plugins/files/tests/plugin.integration.test.ts
+++ b/packages/appkit/src/plugins/files/tests/plugin.integration.test.ts
@@ -438,17 +438,19 @@ describe("Files Plugin Integration", () => {
     });
   });
 
-  describe("OBO Gateway", () => {
-    test("production: rejects requests without user token with 401", async () => {
-      const response = await fetch(`${baseUrl}/api/files/${VOL}/list`);
+  describe("Service principal execution", () => {
+    test("requests without user token return 401 (policy requires user identity)", async () => {
+      // Use a unique path to avoid cached results from earlier tests
+      const response = await fetch(
+        `${baseUrl}/api/files/${VOL}/list?path=sp-only`,
+      );
 
       expect(response.status).toBe(401);
-      const data = (await response.json()) as { error: string; plugin: string };
-      expect(data.plugin).toBe("files");
-      expect(data.error).toMatch(/token/i);
+      const data = (await response.json()) as { error: string };
+      expect(data.error).toMatch(/x-forwarded-user/);
     });
 
-    test("production: allows requests with valid user token (OBO)", async () => {
+    test("requests with user headers also succeed", async () => {
       mockFilesApi.listDirectoryContents.mockReturnValue(
         (async function* () {
           yield {
@@ -466,51 +468,21 @@ describe("Files Plugin Integration", () => {
       expect(response.status).toBe(200);
     });
 
-    test("development: falls back to service principal when no user token", async () => {
-      const originalEnv = process.env.NODE_ENV;
-      process.env.NODE_ENV = "development";
-
-      try {
-        mockFilesApi.listDirectoryContents.mockReturnValue(
-          (async function* () {
-            yield {
-              name: "dev-file.txt",
-              path: "/Volumes/catalog/schema/vol/dev-file.txt",
-              is_directory: false,
-            };
-          })(),
-        );
-
-        const response = await fetch(`${baseUrl}/api/files/${VOL}/list`);
+    test("write operations without explicit policy are denied by default publicRead()", async () => {
+      mockFilesApi.createDirectory.mockResolvedValue(undefined);
 
-        expect(response.status).toBe(200);
-        const data = await response.json();
-        expect(data).toEqual([
-          {
-            name: "dev-file.txt",
-            path: "/Volumes/catalog/schema/vol/dev-file.txt",
-            is_directory: false,
-          },
-        ]);
-      } finally {
-        if (originalEnv === undefined) {
-          delete process.env.NODE_ENV;
-        } else {
-          process.env.NODE_ENV = originalEnv;
-        }
-      }
-    });
-
-    test("production: rejects write operations without user token", async () => {
       const response = await fetch(`${baseUrl}/api/files/${VOL}/mkdir`, {
         method: "POST",
-        headers: { "Content-Type": "application/json" },
-        body: JSON.stringify({ path: "/Volumes/catalog/schema/vol/newdir" }),
+        headers: {
+          ...MOCK_AUTH_HEADERS,
+          "Content-Type": "application/json",
+        },
+        body: JSON.stringify({ path: "newdir" }),
       });
 
-      expect(response.status).toBe(401);
-      const data = (await response.json()) as { error: string; plugin: string };
-      expect(data.plugin).toBe("files");
+      expect(response.status).toBe(403);
+      const data = (await response.json()) as { error: string };
+      expect(data.error).toMatch(/Policy denied/);
     });
   });
 
@@ -607,22 +579,21 @@ describe("Files Plugin Integration", () => {
     });
 
     test("ApiError 409 is swallowed and returns 500", async () => {
-      mockFilesApi.createDirectory.mockRejectedValue(
+      mockFilesApi.getMetadata.mockRejectedValue(
         new MockApiError("Conflict", 409),
       );
 
-      const response = await fetch(`${baseUrl}/api/files/${VOL}/mkdir`, {
-        method: "POST",
-        headers: { ...MOCK_AUTH_HEADERS, "Content-Type": "application/json" },
-        body: JSON.stringify({ path: "/Volumes/catalog/schema/vol/existing" }),
-      });
+      const response = await fetch(
+        `${baseUrl}/api/files/${VOL}/metadata?path=/Volumes/catalog/schema/vol/existing`,
+        { headers: MOCK_AUTH_HEADERS },
+      );
 
       expect(response.status).toBe(500);
       const data = (await response.json()) as {
         error: string;
         plugin: string;
       };
-      expect(data.error).toBe("Create directory failed");
+      expect(data.error).toBe("Metadata fetch failed");
       expect(data.plugin).toBe("files");
     });
   });
diff --git a/packages/appkit/src/plugins/files/tests/plugin.test.ts b/packages/appkit/src/plugins/files/tests/plugin.test.ts
index 27b55cd5..bd92d56b 100644
--- a/packages/appkit/src/plugins/files/tests/plugin.test.ts
+++ b/packages/appkit/src/plugins/files/tests/plugin.test.ts
@@ -9,6 +9,7 @@ import {
   FILES_WRITE_DEFAULTS,
 } from "../defaults";
 import { FilesPlugin, files } from "../plugin";
+import { PolicyDeniedError, policy } from "../policy";
 
 const { mockClient, MockApiError, mockCacheInstance } = vi.hoisted(() => {
   const mockFilesApi = {
@@ -60,7 +61,7 @@ vi.mock("../../../context", async (importOriginal) => {
   return {
     ...actual,
     getWorkspaceClient: vi.fn(() => mockClient),
-    isInUserContext: vi.fn(() => true),
+    getCurrentUserId: vi.fn(() => "test-service-principal"),
   };
 });
 
@@ -72,8 +73,8 @@ vi.mock("../../../cache", () => ({
 
 const VOLUMES_CONFIG = {
   volumes: {
-    uploads: { maxUploadSize: 100_000_000 },
-    exports: {},
+    uploads: { maxUploadSize: 100_000_000, policy: policy.allowAll() },
+    exports: { policy: policy.allowAll() },
   },
 };
 
@@ -247,7 +248,7 @@ describe("FilesPlugin", () => {
     });
   });
 
-  describe("OBO and service principal access", () => {
+  describe("Service principal access", () => {
     const volumeMethods = [
       "list",
       "read",
@@ -270,7 +271,7 @@ describe("FilesPlugin", () => {
       }
     });
 
-    test("asUser throws AuthenticationError without token in production", () => {
+    test("asUser without user header in production → throws AuthenticationError", () => {
       const originalEnv = process.env.NODE_ENV;
       process.env.NODE_ENV = "production";
 
@@ -292,7 +293,10 @@ describe("FilesPlugin", () => {
       try {
         const plugin = new FilesPlugin(VOLUMES_CONFIG);
         const handle = plugin.exports()("uploads");
-        const mockReq = { header: () => undefined } as any;
+        const mockReq = {
+          header: (name: string) =>
+            name === "x-forwarded-user" ? "test-user" : undefined,
+        } as any;
         const api = handle.asUser(mockReq);
 
         for (const method of volumeMethods) {
@@ -303,17 +307,12 @@ describe("FilesPlugin", () => {
       }
     });
 
-    test("direct methods on handle throw without user context (OBO enforced)", async () => {
-      const { isInUserContext } = await import("../../../context");
-      (isInUserContext as ReturnType<typeof vi.fn>).mockReturnValue(false);
-
+    test("direct methods on handle work as service principal", () => {
       const plugin = new FilesPlugin(VOLUMES_CONFIG);
       const handle = plugin.exports()("uploads");
 
-      // Direct call without user context should throw synchronously
-      expect(() => handle.list()).toThrow(
-        'app.files("uploads").list() called without user context (service principal). Use OBO instead: app.files("uploads").asUser(req).list()',
-      );
+      // Direct call executes as service principal (returns a promise, does not throw)
+      expect(typeof handle.list).toBe("function");
     });
   });
 
@@ -493,11 +492,16 @@ describe("FilesPlugin", () => {
       const handler = getUploadHandler(plugin);
       const res = mockRes();
 
+      const headers: Record<string, string> = {
+        "content-length": String(100_000_000),
+        "x-forwarded-user": "test-user",
+      };
       await handler(
         {
           params: { volumeKey: "uploads" },
           query: { path: "/file.bin" },
-          headers: { "content-length": String(100_000_000) },
+          headers,
+          header: (name: string) => headers[name.toLowerCase()],
         },
         res,
       );
@@ -515,11 +519,15 @@ describe("FilesPlugin", () => {
       const handler = getUploadHandler(plugin);
       const res = mockRes();
 
+      const headers: Record<string, string> = {
+        "x-forwarded-user": "test-user",
+      };
       await handler(
         {
           params: { volumeKey: "uploads" },
           query: { path: "/file.bin" },
-          headers: {},
+          headers,
+          header: (name: string) => headers[name.toLowerCase()],
         },
         res,
       );
@@ -644,6 +652,8 @@ describe("FilesPlugin", () => {
 
       const handlerPromise = handler(mockReq("uploads"), res);
 
+      // Flush microtasks (policy check) before advancing timers
+      await vi.advanceTimersByTimeAsync(0);
       await vi.advanceTimersByTimeAsync(100);
       await handlerPromise;
 
@@ -869,6 +879,8 @@ describe("FilesPlugin", () => {
 
       const handlerPromise = handler(mockReq("uploads"), res);
 
+      // Flush microtasks (policy check) before advancing timers
+      await vi.advanceTimersByTimeAsync(0);
       // Advance past read-tier timeout (30s)
       await vi.advanceTimersByTimeAsync(31_000);
       await handlerPromise;
@@ -893,6 +905,376 @@ describe("FilesPlugin", () => {
     });
   });
 
+  describe("Policy enforcement", () => {
+    const POLICY_CONFIG = {
+      volumes: {
+        public: { policy: policy.publicRead() },
+        locked: { policy: policy.denyAll() },
+        open: { policy: policy.allowAll() },
+        uploads: {},
+        exports: {},
+      },
+    };
+
+    function getRouteHandler(
+      plugin: FilesPlugin,
+      method: "get" | "post" | "delete",
+      pathSuffix: string,
+    ) {
+      const mockRouter = {
+        use: vi.fn(),
+        get: vi.fn(),
+        post: vi.fn(),
+        put: vi.fn(),
+        delete: vi.fn(),
+        patch: vi.fn(),
+      } as any;
+      plugin.injectRoutes(mockRouter);
+      const call = mockRouter[method].mock.calls.find(
+        (c: unknown[]) =>
+          typeof c[0] === "string" && (c[0] as string).endsWith(pathSuffix),
+      );
+      return call[call.length - 1] as (req: any, res: any) => Promise<void>;
+    }
+
+    function mockRes() {
+      const res: any = { headersSent: false };
+      res.status = vi.fn().mockReturnValue(res);
+      res.json = vi.fn().mockReturnValue(res);
+      res.type = vi.fn().mockReturnValue(res);
+      res.send = vi.fn().mockReturnValue(res);
+      res.setHeader = vi.fn().mockReturnValue(res);
+      res.end = vi.fn();
+      return res;
+    }
+
+    function mockReq(volumeKey: string, overrides: Record<string, any> = {}) {
+      const headers: Record<string, string> = {
+        "x-forwarded-access-token": "test-token",
+        "x-forwarded-user": "test-user",
+        ...(overrides.headers ?? {}),
+      };
+      return {
+        params: { volumeKey },
+        query: {},
+        ...overrides,
+        headers,
+        header: (name: string) => headers[name.toLowerCase()],
+      };
+    }
+
+    beforeEach(() => {
+      process.env.DATABRICKS_VOLUME_PUBLIC = "/Volumes/c/s/public";
+      process.env.DATABRICKS_VOLUME_LOCKED = "/Volumes/c/s/locked";
+      process.env.DATABRICKS_VOLUME_OPEN = "/Volumes/c/s/open";
+    });
+
+    afterEach(() => {
+      delete process.env.DATABRICKS_VOLUME_PUBLIC;
+      delete process.env.DATABRICKS_VOLUME_LOCKED;
+      delete process.env.DATABRICKS_VOLUME_OPEN;
+    });
+
+    test("policy volume + no user header (production) → 401", async () => {
+      const originalEnv = process.env.NODE_ENV;
+      process.env.NODE_ENV = "production";
+      try {
+        const plugin = new FilesPlugin(POLICY_CONFIG);
+        const handler = getRouteHandler(plugin, "get", "/list");
+        const res = mockRes();
+
+        // Override both headers to undefined so _extractUser has no user
+        const noUserHeaders: Record<string, string> = {};
+        await handler(
+          {
+            params: { volumeKey: "public" },
+            query: {},
+            headers: noUserHeaders,
+            header: (name: string) => noUserHeaders[name.toLowerCase()],
+          },
+          res,
+        );
+
+        expect(res.status).toHaveBeenCalledWith(401);
+      } finally {
+        process.env.NODE_ENV = originalEnv;
+      }
+    });
+
+    test("policy volume + policy returns false → 403", async () => {
+      const plugin = new FilesPlugin(POLICY_CONFIG);
+      const handler = getRouteHandler(plugin, "get", "/list");
+      const res = mockRes();
+
+      await handler(mockReq("locked"), res);
+
+      expect(res.status).toHaveBeenCalledWith(403);
+      expect(res.json).toHaveBeenCalledWith(
+        expect.objectContaining({
+          error: expect.stringContaining("Policy denied"),
+        }),
+      );
+    });
+
+    test("policy volume + policy returns true → 200, runs as SP", async () => {
+      const plugin = new FilesPlugin(POLICY_CONFIG);
+      const handler = getRouteHandler(plugin, "get", "/list");
+      const res = mockRes();
+
+      mockClient.files.listDirectoryContents.mockImplementation(
+        async function* () {
+          yield { name: "a.txt", path: "/a.txt", is_directory: false };
+        },
+      );
+
+      await handler(mockReq("public", { query: {} }), res);
+
+      expect(res.json).toHaveBeenCalledWith(
+        expect.arrayContaining([expect.objectContaining({ name: "a.txt" })]),
+      );
+      // Should NOT have received a 401 or 403
+      const statusCodes = (res.status.mock.calls as number[][]).map(
+        (c) => c[0],
+      );
+      expect(statusCodes).not.toContain(401);
+      expect(statusCodes).not.toContain(403);
+    });
+
+    test("policy volume + async policy → works", async () => {
+      const asyncConfig = {
+        volumes: {
+          async_vol: {
+            policy: async () => true,
+          },
+          uploads: {},
+          exports: {},
+        },
+      };
+      process.env.DATABRICKS_VOLUME_ASYNC_VOL = "/Volumes/c/s/async";
+
+      try {
+        const plugin = new FilesPlugin(asyncConfig);
+        const handler = getRouteHandler(plugin, "get", "/list");
+        const res = mockRes();
+
+        mockClient.files.listDirectoryContents.mockImplementation(
+          async function* () {
+            yield { name: "b.txt", path: "/b.txt", is_directory: false };
+          },
+        );
+
+        await handler(mockReq("async_vol"), res);
+
+        expect(res.json).toHaveBeenCalledWith(
+          expect.arrayContaining([expect.objectContaining({ name: "b.txt" })]),
+        );
+      } finally {
+        delete process.env.DATABRICKS_VOLUME_ASYNC_VOL;
+      }
+    });
+
+    test("default publicRead() volume → reads succeed", async () => {
+      const plugin = new FilesPlugin(POLICY_CONFIG);
+      const handler = getRouteHandler(plugin, "get", "/list");
+      const res = mockRes();
+
+      mockClient.files.listDirectoryContents.mockImplementation(
+        async function* () {
+          yield { name: "c.txt", path: "/c.txt", is_directory: false };
+        },
+      );
+
+      await handler(mockReq("uploads"), res);
+
+      // Should succeed (reads allowed by publicRead default)
+      expect(res.json).toHaveBeenCalledWith(
+        expect.arrayContaining([expect.objectContaining({ name: "c.txt" })]),
+      );
+    });
+
+    test("default publicRead() volume → writes denied with 403", async () => {
+      const plugin = new FilesPlugin(POLICY_CONFIG);
+      const handler = getRouteHandler(plugin, "post", "/mkdir");
+      const res = mockRes();
+
+      await handler(
+        mockReq("uploads", {
+          body: { path: "/newdir" },
+        }),
+        res,
+      );
+
+      expect(res.status).toHaveBeenCalledWith(403);
+      expect(res.json).toHaveBeenCalledWith(
+        expect.objectContaining({
+          error: expect.stringContaining("Policy denied"),
+        }),
+      );
+    });
+
+    test("upload with policy → policy receives size in resource", async () => {
+      const policySpy = vi.fn().mockReturnValue(true);
+      const sizeConfig = {
+        volumes: {
+          sized: { policy: policySpy },
+          uploads: {},
+          exports: {},
+        },
+      };
+      process.env.DATABRICKS_VOLUME_SIZED = "/Volumes/c/s/sized";
+
+      try {
+        const plugin = new FilesPlugin(sizeConfig);
+        const handler = getRouteHandler(plugin, "post", "/upload");
+        const res = mockRes();
+
+        await handler(
+          mockReq("sized", {
+            query: { path: "/test.bin" },
+            headers: {
+              "content-length": "12345",
+              "x-forwarded-user": "test-user",
+              "x-forwarded-access-token": "test-token",
+            },
+          }),
+          res,
+        );
+
+        expect(policySpy).toHaveBeenCalledWith(
+          "upload",
+          expect.objectContaining({ size: 12345 }),
+          expect.objectContaining({ id: "test-user" }),
+        );
+      } finally {
+        delete process.env.DATABRICKS_VOLUME_SIZED;
+      }
+    });
+
+    test("SDK asUser(req) on policy volume → policy-wrapped API works", async () => {
+      const plugin = new FilesPlugin(POLICY_CONFIG);
+      const exported = plugin.exports();
+      const handle = exported("public");
+
+      mockClient.files.listDirectoryContents.mockImplementation(
+        async function* () {
+          yield { name: "d.txt", path: "/d.txt", is_directory: false };
+        },
+      );
+
+      const mockReqObj = {
+        header: (name: string) => {
+          if (name === "x-forwarded-user") return "test-user";
+          if (name === "x-forwarded-access-token") return "test-token";
+          return undefined;
+        },
+      } as any;
+
+      const api = handle.asUser(mockReqObj);
+      const result = await api.list();
+      expect(result).toEqual(
+        expect.arrayContaining([expect.objectContaining({ name: "d.txt" })]),
+      );
+    });
+
+    test("SDK asUser(req) on policy volume + deny → throws PolicyDeniedError", async () => {
+      const plugin = new FilesPlugin(POLICY_CONFIG);
+      const exported = plugin.exports();
+      const handle = exported("locked");
+
+      const mockReqObj = {
+        header: (name: string) => {
+          if (name === "x-forwarded-user") return "test-user";
+          if (name === "x-forwarded-access-token") return "test-token";
+          return undefined;
+        },
+      } as any;
+
+      const api = handle.asUser(mockReqObj);
+      await expect(api.list()).rejects.toThrow(PolicyDeniedError);
+    });
+
+    test("direct call on policy volume → enforces policy as SP", async () => {
+      const plugin = new FilesPlugin(POLICY_CONFIG);
+      const handle = plugin.exports()("open");
+
+      // Direct call on allowAll() volume succeeds (policy is checked but allows)
+      const result = await handle.list();
+      expect(result).toEqual(
+        expect.arrayContaining([expect.objectContaining({ name: "d.txt" })]),
+      );
+    });
+
+    test("direct SP call on denyAll() volume → throws PolicyDeniedError", async () => {
+      const plugin = new FilesPlugin(POLICY_CONFIG);
+      const handle = plugin.exports()("locked");
+
+      await expect(handle.list()).rejects.toThrow(PolicyDeniedError);
+    });
+
+    test("direct SP call → policy receives { isServicePrincipal: true }", async () => {
+      const policySpy = vi.fn().mockReturnValue(true);
+      const spyConfig = {
+        volumes: {
+          spied: { policy: policySpy },
+          uploads: {},
+          exports: {},
+        },
+      };
+      process.env.DATABRICKS_VOLUME_SPIED = "/Volumes/c/s/spied";
+
+      try {
+        const plugin = new FilesPlugin(spyConfig);
+        const handle = plugin.exports()("spied");
+        await handle.list();
+
+        expect(policySpy).toHaveBeenCalledWith(
+          "list",
+          expect.objectContaining({ volume: "spied" }),
+          expect.objectContaining({ isServicePrincipal: true }),
+        );
+      } finally {
+        delete process.env.DATABRICKS_VOLUME_SPIED;
+      }
+    });
+
+    test("asUser() call → policy receives user without isServicePrincipal", async () => {
+      const policySpy = vi.fn().mockReturnValue(true);
+      const spyConfig = {
+        volumes: {
+          spied: { policy: policySpy },
+          uploads: {},
+          exports: {},
+        },
+      };
+      process.env.DATABRICKS_VOLUME_SPIED = "/Volumes/c/s/spied";
+
+      try {
+        const plugin = new FilesPlugin(spyConfig);
+        const handle = plugin.exports()("spied");
+        const mockReqObj = {
+          header: (name: string) => {
+            if (name === "x-forwarded-user") return "test-user";
+            if (name === "x-forwarded-access-token") return "test-token";
+            return undefined;
+          },
+        } as any;
+
+        await handle.asUser(mockReqObj).list();
+
+        expect(policySpy).toHaveBeenCalledWith(
+          "list",
+          expect.objectContaining({ volume: "spied" }),
+          expect.objectContaining({ id: "test-user" }),
+        );
+        // Should NOT have isServicePrincipal set
+        const userArg = policySpy.mock.calls[0][2];
+        expect(userArg.isServicePrincipal).toBeUndefined();
+      } finally {
+        delete process.env.DATABRICKS_VOLUME_SPIED;
+      }
+    });
+  });
+
   describe("Upload Stream Size Limiter", () => {
     test("stream under limit passes through all chunks", async () => {
       const maxSize = 100;
diff --git a/packages/appkit/src/plugins/files/tests/policy.test.ts b/packages/appkit/src/plugins/files/tests/policy.test.ts
new file mode 100644
index 00000000..b835245a
--- /dev/null
+++ b/packages/appkit/src/plugins/files/tests/policy.test.ts
@@ -0,0 +1,136 @@
+import { describe, expect, test } from "vitest";
+import {
+  type FileAction,
+  type FilePolicyUser,
+  type FileResource,
+  PolicyDeniedError,
+  policy,
+  READ_ACTIONS,
+  WRITE_ACTIONS,
+} from "../policy";
+
+const user: FilePolicyUser = { id: "user-1" };
+const resource: FileResource = { path: "/file.txt", volume: "uploads" };
+
+describe("FileAction sets", () => {
+  test("READ_ACTIONS contains all read actions", () => {
+    for (const a of [
+      "list",
+      "read",
+      "download",
+      "raw",
+      "exists",
+      "metadata",
+      "preview",
+    ] as FileAction[]) {
+      expect(READ_ACTIONS.has(a)).toBe(true);
+    }
+  });
+
+  test("WRITE_ACTIONS contains all write actions", () => {
+    for (const a of ["upload", "mkdir", "delete"] as FileAction[]) {
+      expect(WRITE_ACTIONS.has(a)).toBe(true);
+    }
+  });
+
+  test("READ_ACTIONS and WRITE_ACTIONS are disjoint", () => {
+    for (const a of READ_ACTIONS) {
+      expect(WRITE_ACTIONS.has(a)).toBe(false);
+    }
+  });
+});
+
+describe("policy.publicRead()", () => {
+  const p = policy.publicRead();
+
+  test("allows read actions", () => {
+    for (const a of READ_ACTIONS) {
+      expect(p(a, resource, user)).toBe(true);
+    }
+  });
+
+  test("denies write actions", () => {
+    for (const a of WRITE_ACTIONS) {
+      expect(p(a, resource, user)).toBe(false);
+    }
+  });
+});
+
+describe("policy.denyAll() / policy.allowAll()", () => {
+  test("denyAll denies everything", () => {
+    const p = policy.denyAll();
+    expect(p("list", resource, user)).toBe(false);
+    expect(p("upload", resource, user)).toBe(false);
+  });
+
+  test("allowAll allows everything", () => {
+    const p = policy.allowAll();
+    expect(p("list", resource, user)).toBe(true);
+    expect(p("upload", resource, user)).toBe(true);
+  });
+});
+
+describe("policy.all()", () => {
+  test("returns true when all policies allow", async () => {
+    const p = policy.all(policy.allowAll(), policy.allowAll());
+    expect(await p("list", resource, user)).toBe(true);
+  });
+
+  test("short-circuits on first deny", async () => {
+    let secondCalled = false;
+    const p = policy.all(policy.denyAll(), () => {
+      secondCalled = true;
+      return true;
+    });
+    expect(await p("list", resource, user)).toBe(false);
+    expect(secondCalled).toBe(false);
+  });
+});
+
+describe("policy.any()", () => {
+  test("returns false when all policies deny", async () => {
+    const p = policy.any(policy.denyAll(), policy.denyAll());
+    expect(await p("list", resource, user)).toBe(false);
+  });
+
+  test("short-circuits on first allow", async () => {
+    let secondCalled = false;
+    const p = policy.any(policy.allowAll(), () => {
+      secondCalled = true;
+      return false;
+    });
+    expect(await p("list", resource, user)).toBe(true);
+    expect(secondCalled).toBe(false);
+  });
+});
+
+describe("policy.not()", () => {
+  test("inverts allow to deny", async () => {
+    const p = policy.not(policy.allowAll());
+    expect(await p("list", resource, user)).toBe(false);
+  });
+
+  test("inverts deny to allow", async () => {
+    const p = policy.not(policy.denyAll());
+    expect(await p("list", resource, user)).toBe(true);
+  });
+});
+
+describe("async policy support", () => {
+  test("handles async policy that returns Promise<boolean>", async () => {
+    const asyncPolicy = async () => true;
+    const p = policy.all(asyncPolicy);
+    expect(await p("list", resource, user)).toBe(true);
+  });
+});
+
+describe("PolicyDeniedError", () => {
+  test("has correct name and message", () => {
+    const err = new PolicyDeniedError("upload", "images");
+    expect(err.name).toBe("PolicyDeniedError");
+    expect(err.message).toBe('Policy denied "upload" on volume "images"');
+    expect(err.action).toBe("upload");
+    expect(err.volumeKey).toBe("images");
+    expect(err instanceof Error).toBe(true);
+  });
+});
diff --git a/packages/appkit/src/plugins/files/types.ts b/packages/appkit/src/plugins/files/types.ts
index 9a37a4a1..82b54688 100644
--- a/packages/appkit/src/plugins/files/types.ts
+++ b/packages/appkit/src/plugins/files/types.ts
@@ -1,5 +1,6 @@
 import type { files } from "@databricks/sdk-experimental";
 import type { BasePluginConfig, IAppRequest } from "shared";
+import type { FilePolicy } from "./policy";
 
 /**
  * Per-volume configuration options.
@@ -9,11 +10,17 @@ export interface VolumeConfig {
   maxUploadSize?: number;
   /** Map of file extensions to MIME types for this volume. Inherits from plugin-level `customContentTypes` if not set. */
   customContentTypes?: Record<string, string>;
+  /**
+   * Access-control policy for this volume. When set, operations execute as the
+   * service principal and the policy decides whether the action is allowed.
+   */
+  policy?: FilePolicy;
 }
 
 /**
  * User-facing API for a single volume.
- * Prefer OBO access via `app.files("volumeKey").asUser(req).list()`.
+ * All operations execute as the service principal. When a policy is
+ * configured on the volume, every call is checked against that policy.
  */
 export interface VolumeAPI {
   list(directoryPath?: string): Promise<DirectoryEntry[]>;
@@ -78,8 +85,9 @@ export interface FilePreview extends FileMetadata {
 /**
  * Volume handle returned by `app.files("volumeKey")`.
  *
- * - `asUser(req)` — executes on behalf of the user (recommended).
- * - Direct methods (e.g. `.list()`) — execute as the service principal (logs a warning encouraging OBO).
+ * All methods execute as the service principal and enforce the volume's
+ * policy (if configured) with `{ isServicePrincipal: true }`.
+ * `asUser(req)` re-wraps with the real user identity for per-user policy checks.
  */
 export type VolumeHandle = VolumeAPI & {
   asUser: (req: IAppRequest) => VolumeAPI;
@@ -91,15 +99,15 @@ export type VolumeHandle = VolumeAPI & {
  *
  * @example
  * ```ts
- * // OBO access (recommended)
- * appKit.files("uploads").asUser(req).list()
- *
- * // Service principal access (logs a warning)
+ * // Service principal access
  * appKit.files("uploads").list()
  *
+ * // With policy: pass user identity for access control
+ * appKit.files("uploads").asUser(req).list()
+ *
  * // Named accessor
  * const vol = appKit.files.volume("uploads")
- * await vol.asUser(req).list()
+ * await vol.list()
  * ```
  */
 export interface FilesExport {