feat(appkit): add on-behalf-of-user mode for agents

docs/docs/plugins/agents.md

@@ -317,6 +317,47 @@ const result = await runAgent(classifier, {
 
 MCP hosted tools (`mcpServer(...)`) still require `agents()` (they need a live MCP client). Supervisor-API hosted tools (`supervisorTools.*`), by contrast, **work in standalone `runAgent`** — the adapter has everything it needs to execute them server-side. This makes batch-eval / CI use of supervisor agents possible without `createApp`. Plugin tool dispatch in standalone mode runs as the service principal (no OBO) and **bypasses the agents-plugin approval gate** — treat standalone runAgent as a trusted-prompt environment (CI, batch eval, internal scripts), not as an exposed user-facing surface.
 
+## Execution identity
+
+By default an agent runs **mixed**: the model call and hand-rolled `tool({ execute })` tools run as the app's service principal, and plugin-toolkit tools run as the requesting user. Set `auth: "on-behalf-of-user"` to run the whole agent as the user:
+
+```ts
+agents({ auth: "on-behalf-of-user" });                      // default for every agent
+createAgent({ instructions: "...", auth: "on-behalf-of-user" }); // one agent
+```
+
+In markdown, set `auth: on-behalf-of-user` in the frontmatter. A per-agent value overrides the plugin default. Omitting `auth` keeps the mixed behavior; there is no all-service-principal mode.
+
+| Piece | Default (mixed) | `on-behalf-of-user` |
+|---|---|---|
+| Model call | service principal | user |
+| Plugin-toolkit tools | user | user |
+| Hand-rolled `tool({ execute })` | service principal | user |
+| Sub-agents | own mode | own mode, never service principal |
+| Standalone `runAgent` | service principal, or user tools with `caller` | requires `caller` (or an ambient user scope); model and tools as user |
+| MLflow tracing | service principal | service principal (exception) |
+| Thread store | service principal | service principal (exception) |
+| Catalog skills volume | service principal | service principal (exception) |
+| Missing user token | plugin tools reject; the rest runs as the service principal | the request is rejected with 401 before any model or tool call |
+
+An on-behalf-of-user agent fails closed:
+
+- No forwarded user token: `401` before any model or tool call, in production and in development. There is no service-principal fallback.
+- A `401` from the model mid-run becomes an `IDENTITY_EXPIRED` error event and the stream ends. The run is not retried as the service principal.
+- A sub-agent never widens: under an on-behalf-of-user parent, a mixed sub-agent also runs as the user.
+
+**Exceptions.** MLflow tracing, the thread store, and catalog skills (see [Catalog skills](#catalog-skills-unity-catalog-volume)) are app-owned and stay service principal in every mode. Thread rows are keyed by the user id.
+
+**Pre-built adapters.** The user's client is applied when AppKit builds the adapter from a model string (`model: "my-endpoint"`, `defaultModel`, or `DATABRICKS_SERVING_ENDPOINT_NAME`). If you build a `DatabricksAdapter` yourself with a fixed `workspaceClient`, an on-behalf-of-user agent throws at boot instead of running the model as the service principal. Pass a provider so the client resolves per call, or use a model string. Mixed agents accept fixed-client adapters as before:
+
+```ts
+DatabricksAdapter.fromModelServing("my-endpoint", {
+  workspaceClient: () => getWorkspaceClient(),
+});
+```
+
+**Provisioning.** Each user needs the `model-serving` user API scope on the app, and `CAN_QUERY` on any custom serving endpoint the agent calls. Plugin tools still need their own scopes and grants as in mixed mode.
+
 ## Adding agents to an existing app
 
 Already have an app and want to add agents? What you touch depends on the kind:
@@ -477,6 +518,7 @@ agents({
   agents?: Record<string, AgentDefinition>,  // DEPRECATED — use server/agents/<id>/ discovery
   defaultAgent?: string,
   defaultModel?: AgentAdapter | Promise<AgentAdapter> | string,
+  auth?: "on-behalf-of-user",  // default: mixed (see Execution identity)
   tools?: Record<string, AgentTool>,
   autoInheritTools?: boolean | { file?: boolean, code?: boolean },
   autoInheritSkills?: boolean | { file?: boolean, code?: boolean }, // default off
@@ -916,6 +958,7 @@ Skip `--experiment` (and `MLFLOW_EXPERIMENT_ID`) to run evals purely locally wit
 | `maxTokens` | number | Adapter max-token hint. |
 | `generationParams` | object | Adapter generation params (e.g. `temperature`, `top_p`) passed through when AppKit builds the adapter. |
 | `baseSystemPrompt` | false \| string | Per-agent override. `false` disables the AppKit base prompt. |
+| `auth` | string | `on-behalf-of-user` runs this agent as the user. Any other value throws at boot. See [Execution identity](#execution-identity). |
 | `ephemeral` | boolean | If `true`, the thread created for a chat request against this agent is deleted from `ThreadStore` after the stream finishes. Use for stateless one-shot agents (e.g. autocomplete) so history does not accumulate or contaminate future calls. Defaults to `false`. |
 
 Unknown keys are logged and ignored. Invalid YAML and missing plugin/tool references throw at boot.

packages/appkit/src/agents/databricks.ts

@@ -180,14 +180,46 @@ function isStreamBodyOptions(
  * interface rather than importing the SDK type directly. This keeps the adapter
  * free of a hard compile-time dependency on `@databricks/sdk-experimental`.
  */
-interface WorkspaceClientLike {
+export interface WorkspaceClientLike {
   apiClient: {
     request(options: Record<string, unknown>): Promise<unknown>;
   };
 }
 
+/**
+ * A fixed client, or a provider resolved on every model call. A provider lets
+ * one adapter follow the active execution scope (for example
+ * `() => getWorkspaceClient()` for on-behalf-of-user agents).
+ */
+type WorkspaceClientSource = WorkspaceClientLike | (() => WorkspaceClientLike);
+
+function clientResolver(
+  source: WorkspaceClientSource,
+): () => WorkspaceClientLike {
+  return typeof source === "function" ? source : () => source;
+}
+
+const fixedClientAdapters = new WeakSet<AgentAdapter>();
+
+/** Record whether an adapter's client is fixed (not a per-call provider). */
+function withClientSource<T extends AgentAdapter>(
+  source: WorkspaceClientSource | undefined,
+  adapter: T,
+): T {
+  if (typeof source !== "function") fixedClientAdapters.add(adapter);
+  return adapter;
+}
+
+/**
+ * @internal True when a Databricks adapter was built with a fixed workspace
+ * client, so it cannot follow the caller in an on-behalf-of-user agent.
+ */
+export function hasFixedWorkspaceClient(adapter: AgentAdapter): boolean {
+  return fixedClientAdapters.has(adapter);
+}
+
 interface ServingEndpointOptions {
-  workspaceClient: WorkspaceClientLike;
+  workspaceClient: WorkspaceClientSource;
   endpointName: string;
   maxSteps?: number;
   maxTokens?: number;
@@ -201,7 +233,7 @@ interface ModelServingOptions {
   maxSteps?: number;
   maxTokens?: number;
   generationParams?: GenerationParams;
-  workspaceClient?: WorkspaceClientLike;
+  workspaceClient?: WorkspaceClientSource;
   maxSseLineChars?: number;
   maxStreamTextChars?: number;
   maxToolArgumentsChars?: number;
@@ -218,8 +250,10 @@ interface AiGatewayOptions {
    * is created from the ambient client options (SDK credential chain). It is
    * captured once and reused across requests — do not pass a per-request OBO
    * client (it would leak the first request's identity into later ones).
+   * To follow the caller per request, pass a provider such as
+   * `() => getWorkspaceClient()`; it is resolved on every model call.
    */
-  workspaceClient?: WorkspaceClientLike;
+  workspaceClient?: WorkspaceClientSource;
   maxSteps?: number;
   maxTokens?: number;
   generationParams?: GenerationParams;
@@ -386,13 +420,14 @@ export class DatabricksAdapter implements AgentAdapter {
       maxStreamTextChars,
       maxToolArgumentsChars,
     } = options;
-    return new DatabricksAdapter({
+    const resolveClient = clientResolver(workspaceClient);
+    const adapter = new DatabricksAdapter({
       streamBody: (body, signal) =>
         // Cast through the structural shape: the connector types
         // `workspaceClient` as the SDK's concrete `WorkspaceClient`, but we
         // only need `apiClient.request`.
         servingStream(
-          workspaceClient as unknown as Parameters<typeof servingStream>[0],
+          resolveClient() as unknown as Parameters<typeof servingStream>[0],
           endpointName,
           body,
           signal,
@@ -404,6 +439,7 @@ export class DatabricksAdapter implements AgentAdapter {
       maxStreamTextChars,
       maxToolArgumentsChars,
     });
+    return withClientSource(workspaceClient, adapter);
   }
 
   /**
@@ -440,7 +476,7 @@ export class DatabricksAdapter implements AgentAdapter {
       );
     }
 
-    let workspaceClient: WorkspaceClientLike | undefined =
+    let workspaceClient: WorkspaceClientSource | undefined =
       options?.workspaceClient;
     if (!workspaceClient) {
       workspaceClient = createWorkspaceClient({
@@ -508,19 +544,20 @@ export class DatabricksAdapter implements AgentAdapter {
       maxToolArgumentsChars,
     } = options;
 
-    const client =
+    const resolveClient = clientResolver(
       workspaceClient ??
-      (createWorkspaceClient({
-        clientOptions: getClientOptions(),
-      }) as unknown as WorkspaceClientLike);
+        (createWorkspaceClient({
+          clientOptions: getClientOptions(),
+        }) as unknown as WorkspaceClientLike),
+    );
 
-    return new DatabricksAdapter({
+    const adapter = new DatabricksAdapter({
       streamBody: (body, signal) =>
         // Same structural cast as `fromServingEndpoint`: the connector types
         // the client as the SDK's `WorkspaceClient`, but we only need
         // `apiClient.request`.
         streamAiGateway(
-          client as unknown as Parameters<typeof streamAiGateway>[0],
+          resolveClient() as unknown as Parameters<typeof streamAiGateway>[0],
           body,
           signal,
         ),
@@ -532,6 +569,7 @@ export class DatabricksAdapter implements AgentAdapter {
       maxStreamTextChars,
       maxToolArgumentsChars,
     });
+    return withClientSource(workspaceClient, adapter);
   }
 
   /**
@@ -961,7 +999,7 @@ export class DatabricksAdapter implements AgentAdapter {
  */
 type ModelStringOptions = Pick<
   AiGatewayOptions,
-  "maxSteps" | "maxTokens" | "generationParams"
+  "maxSteps" | "maxTokens" | "generationParams" | "workspaceClient"
 >;
 
 /**

packages/appkit/src/agents/tests/databricks.test.ts

@@ -1091,6 +1091,46 @@ describe("DatabricksAdapter", () => {
 });
 
 describe("DatabricksAdapter.fromServingEndpoint", () => {
+  test.each([
+    [
+      "fromServingEndpoint",
+      (workspaceClient: () => { apiClient: unknown }) =>
+        DatabricksAdapter.fromServingEndpoint({
+          workspaceClient: workspaceClient as never,
+          endpointName: "my-model",
+        }),
+    ],
+    [
+      "fromAiGateway",
+      (workspaceClient: () => { apiClient: unknown }) =>
+        DatabricksAdapter.fromAiGateway({
+          workspaceClient: workspaceClient as never,
+          model: "system.ai.claude",
+        }),
+    ],
+  ])("%s resolves a client provider on every run", async (_name, build) => {
+    const clients = ["alice", "bob"].map((user) => ({
+      user,
+      apiClient: {
+        request: vi.fn(async () => ({
+          contents: createReadableStream([textDelta(user), sseChunk("[DONE]")]),
+        })),
+      },
+    }));
+    let next = 0;
+    const adapter = await build(() => clients[next++]);
+    for (const _ of clients) {
+      for await (const _event of adapter.run(
+        { messages: createTestMessages(), tools: [], threadId: "t1" },
+        { executeTool: vi.fn() },
+      )) {
+        // drain
+      }
+    }
+    expect(clients[0].apiClient.request).toHaveBeenCalledTimes(1);
+    expect(clients[1].apiClient.request).toHaveBeenCalledTimes(1);
+  });
+
   test("routes tool-free chat through apiClient.request with a streaming payload", async () => {
     const apiClient = {
       request: vi.fn().mockResolvedValue({

packages/appkit/src/beta.ts

@@ -87,6 +87,7 @@ export {
 export * from "./evals";
 // Agent types
 export type {
+  AgentAuth,
   AgentDefinition,
   AgentsPluginConfig,
   AgentTool,

packages/appkit/src/context/execution-context.ts

@@ -194,6 +194,15 @@ export function isInUserContext(): boolean {
   return ctx !== undefined;
 }
 
+/**
+ * @internal Run `fn` with no caller scope, so it executes as the app service
+ * principal even inside an on-behalf-of-user run. Used for app-owned
+ * state such as the agents thread store.
+ */
+export function runOutsideCallerScope<T>(fn: () => T): T {
+  return executionContextStorage.exit(fn);
+}
+
 /**
  * Get the caller context if one is active, otherwise `undefined`.
  * Unlike `getExecutionContext()`, this does not require `ServiceContext`