diff --git a/apps/cloud/drizzle/0020_add_connection_identity_override.sql b/apps/cloud/drizzle/0020_add_connection_identity_override.sql
new file mode 100644
index 000000000..f1891e3f2
--- /dev/null
+++ b/apps/cloud/drizzle/0020_add_connection_identity_override.sql
@@ -0,0 +1 @@
+ALTER TABLE "connection" ADD COLUMN IF NOT EXISTS "identity_override" json;
diff --git a/apps/cloud/drizzle/meta/_journal.json b/apps/cloud/drizzle/meta/_journal.json
index 04b1d990c..6f1cee27d 100644
--- a/apps/cloud/drizzle/meta/_journal.json
+++ b/apps/cloud/drizzle/meta/_journal.json
@@ -141,6 +141,13 @@
       "when": 1780081200000,
       "tag": "0019_workos_vault_plugin_storage",
       "breakpoints": true
+    },
+    {
+      "idx": 20,
+      "version": "7",
+      "when": 1780124534904,
+      "tag": "0020_add_connection_identity_override",
+      "breakpoints": true
     }
   ]
 }
diff --git a/apps/cloud/src/services/db.schema.test.ts b/apps/cloud/src/services/db.schema.test.ts
index 144a1401e..0ba950628 100644
--- a/apps/cloud/src/services/db.schema.test.ts
+++ b/apps/cloud/src/services/db.schema.test.ts
@@ -18,9 +18,36 @@ import { drizzle } from "drizzle-orm/postgres-js";
 import { Effect } from "effect";
 import postgres from "postgres";
 
+import { collectTables } from "@executor-js/sdk";
+
+import executorConfig from "../../executor.config";
 import * as cloudSchema from "./schema";
 import * as executorSchema from "./executor-schema";
 import { combinedSchema } from "./db";
+import { createDrizzleFumaDb } from "./fuma";
+
+interface GeneratedFumaTable {
+  readonly names: {
+    readonly drizzle: string;
+  };
+  readonly columns: Record<
+    string,
+    {
+      readonly ormName: string;
+      readonly names: {
+        readonly drizzle: string;
+      };
+    }
+  >;
+}
+
+const fumaDbInternals = (
+  value: unknown,
+): { internal: { tables: Record<string, GeneratedFumaTable> } } =>
+  value as { internal: { tables: Record<string, GeneratedFumaTable> } };
+
+const drizzleTableColumns = (value: unknown): Record<string, unknown> =>
+  value as Record<string, unknown>;
 
 describe("combinedSchema", () => {
   it("spreads every cloud + executor schema export", () => {
@@ -64,4 +91,38 @@ describe("combinedSchema", () => {
       ),
     ),
   );
+
+  it.effect("generated drizzle tables expose every executor Fuma column", () =>
+    Effect.acquireRelease(
+      Effect.sync(() => postgres("postgres://u:p@127.0.0.1:1/x", { max: 1 })),
+      (sql) => Effect.promise(() => sql.end({ timeout: 0 })),
+    ).pipe(
+      Effect.flatMap((sql) =>
+        Effect.sync(() => {
+          const db = drizzle(sql, { schema: combinedSchema });
+          const fuma = createDrizzleFumaDb({
+            db,
+            tables: collectTables(executorConfig.plugins({})),
+            namespace: "executor_cloud",
+            provider: "postgresql",
+          });
+          const schemaTables = drizzleTableColumns(combinedSchema);
+          const missingColumns: string[] = [];
+
+          for (const table of Object.values(fumaDbInternals(fuma.db).internal.tables)) {
+            const drizzleTable = drizzleTableColumns(schemaTables[table.names.drizzle]);
+            for (const column of Object.values(table.columns)) {
+              if (drizzleTable[column.names.drizzle] === undefined) {
+                missingColumns.push(
+                  `${table.names.drizzle}.${column.ormName} -> ${column.names.drizzle}`,
+                );
+              }
+            }
+          }
+
+          expect(missingColumns).toEqual([]);
+        }),
+      ),
+    ),
+  );
 });
diff --git a/apps/cloud/src/services/executor-schema.ts b/apps/cloud/src/services/executor-schema.ts
index 69cd41a18..57beba20f 100644
--- a/apps/cloud/src/services/executor-schema.ts
+++ b/apps/cloud/src/services/executor-schema.ts
@@ -1,10 +1,7 @@
-import { pgTable, varchar, text, boolean, timestamp, uniqueIndex, json, bigint } from "drizzle-orm/pg-core"
+import { pgTable, text, boolean, timestamp, varchar, uniqueIndex, json, bigint } from "drizzle-orm/pg-core"
 import { createId } from "fumadb/cuid"
 
 export const source = pgTable("source", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   plugin_id: text("plugin_id").notNull(),
   kind: text("kind").notNull(),
   name: text("name").notNull(),
@@ -13,15 +10,15 @@ export const source = pgTable("source", {
   can_refresh: boolean("can_refresh").notNull().default(false),
   can_edit: boolean("can_edit").notNull().default(false),
   created_at: timestamp("created_at").notNull(),
-  updated_at: timestamp("updated_at").notNull()
+  updated_at: timestamp("updated_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("source_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const tool = pgTable("tool", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   source_id: text("source_id").notNull(),
   plugin_id: text("plugin_id").notNull(),
   name: text("name").notNull(),
@@ -29,40 +26,40 @@ export const tool = pgTable("tool", {
   input_schema: json("input_schema"),
   output_schema: json("output_schema"),
   created_at: timestamp("created_at").notNull(),
-  updated_at: timestamp("updated_at").notNull()
+  updated_at: timestamp("updated_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("tool_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const definition = pgTable("definition", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   source_id: text("source_id").notNull(),
   plugin_id: text("plugin_id").notNull(),
   name: text("name").notNull(),
   schema: json("schema").notNull(),
-  created_at: timestamp("created_at").notNull()
+  created_at: timestamp("created_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("definition_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const secret = pgTable("secret", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   name: text("name").notNull(),
   provider: text("provider").notNull(),
   owned_by_connection_id: text("owned_by_connection_id"),
-  created_at: timestamp("created_at").notNull()
+  created_at: timestamp("created_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("secret_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const connection = pgTable("connection", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   provider: text("provider").notNull(),
   identity_label: text("identity_label"),
   access_token_secret_id: text("access_token_secret_id").notNull(),
@@ -70,16 +67,17 @@ export const connection = pgTable("connection", {
   expires_at: bigint("expires_at", { mode: "bigint" }),
   scope: text("scope"),
   provider_state: json("provider_state"),
+  identity_override: json("identity_override"),
   created_at: timestamp("created_at").notNull(),
-  updated_at: timestamp("updated_at").notNull()
+  updated_at: timestamp("updated_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("connection_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const oauth2_session = pgTable("oauth2_session", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   plugin_id: text("plugin_id").notNull(),
   strategy: text("strategy").notNull(),
   connection_id: text("connection_id").notNull(),
@@ -87,15 +85,15 @@ export const oauth2_session = pgTable("oauth2_session", {
   redirect_url: text("redirect_url").notNull(),
   payload: json("payload").notNull(),
   expires_at: bigint("expires_at", { mode: "bigint" }).notNull(),
-  created_at: timestamp("created_at").notNull()
+  created_at: timestamp("created_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("oauth2_session_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const credential_binding = pgTable("credential_binding", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   plugin_id: text("plugin_id").notNull(),
   source_id: text("source_id").notNull(),
   source_scope_id: text("source_scope_id").notNull(),
@@ -106,44 +104,47 @@ export const credential_binding = pgTable("credential_binding", {
   secret_scope_id: text("secret_scope_id"),
   connection_id: text("connection_id"),
   created_at: timestamp("created_at").notNull(),
-  updated_at: timestamp("updated_at").notNull()
+  updated_at: timestamp("updated_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("credential_binding_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const plugin_storage = pgTable("plugin_storage", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   plugin_id: text("plugin_id").notNull(),
   collection: text("collection").notNull(),
   key: text("key").notNull(),
   data: json("data").notNull(),
   created_at: timestamp("created_at").notNull(),
-  updated_at: timestamp("updated_at").notNull()
+  updated_at: timestamp("updated_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("plugin_storage_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const tool_policy = pgTable("tool_policy", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   pattern: text("pattern").notNull(),
   action: text("action").notNull(),
   position: text("position").notNull(),
   created_at: timestamp("created_at").notNull(),
-  updated_at: timestamp("updated_at").notNull()
+  updated_at: timestamp("updated_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("tool_policy_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const blob = pgTable("blob", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
   namespace: text("namespace").notNull(),
   key: text("key").notNull(),
-  value: text("value").notNull()
+  value: text("value").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("blob_id_uidx").on(table.id)
 ])
@@ -151,4 +152,4 @@ export const blob = pgTable("blob", {
 export const private_executor_cloud_settings = pgTable("private_executor_cloud_settings", {
   id: varchar("id", { length: 255 }).primaryKey().notNull(),
   version: varchar("version", { length: 255 }).notNull().default("1.0.0")
-})
+})
\ No newline at end of file
diff --git a/apps/local/drizzle/0012_connection_identity_override.sql b/apps/local/drizzle/0012_connection_identity_override.sql
new file mode 100644
index 000000000..5d2c612b9
--- /dev/null
+++ b/apps/local/drizzle/0012_connection_identity_override.sql
@@ -0,0 +1 @@
+ALTER TABLE `connection` ADD `identity_override` text;
diff --git a/apps/local/drizzle/meta/_journal.json b/apps/local/drizzle/meta/_journal.json
index 96892e22b..9167ab64d 100644
--- a/apps/local/drizzle/meta/_journal.json
+++ b/apps/local/drizzle/meta/_journal.json
@@ -85,6 +85,13 @@
       "when": 1779087600000,
       "tag": "0011_plugin_storage_sources",
       "breakpoints": true
+    },
+    {
+      "idx": 12,
+      "version": "6",
+      "when": 1779998400000,
+      "tag": "0012_connection_identity_override",
+      "breakpoints": true
     }
   ]
 }
diff --git a/apps/local/src/server/executor-schema.ts b/apps/local/src/server/executor-schema.ts
index 7f36764ba..7e508d44e 100644
--- a/apps/local/src/server/executor-schema.ts
+++ b/apps/local/src/server/executor-schema.ts
@@ -93,6 +93,7 @@ export const connection = sqliteTable(
     expires_at: integer("expires_at"),
     scope: text("scope"),
     provider_state: text("provider_state", { mode: "json" }),
+    identity_override: text("identity_override", { mode: "json" }),
     created_at: integer("created_at", { mode: "timestamp_ms" }).notNull(),
     updated_at: integer("updated_at", { mode: "timestamp_ms" }).notNull(),
   },
diff --git a/apps/local/src/server/sqlite-fumadb.ts b/apps/local/src/server/sqlite-fumadb.ts
index 57fb99d1b..e0bb91a37 100644
--- a/apps/local/src/server/sqlite-fumadb.ts
+++ b/apps/local/src/server/sqlite-fumadb.ts
@@ -53,6 +53,15 @@ export const createSqliteFumaDb = async <const TTables extends FumaTables>(
   })) {
     sqlite.exec(statement);
   }
+  const connectionColumns = sqlite
+    .prepare("PRAGMA table_info('connection')")
+    .all() as ReadonlyArray<{ readonly name: string }>;
+  if (
+    connectionColumns.length > 0 &&
+    !connectionColumns.some((column) => column.name === "identity_override")
+  ) {
+    sqlite.exec("ALTER TABLE connection ADD COLUMN identity_override TEXT");
+  }
 
   const latestSchema = fumaSchema({
     version,
diff --git a/packages/core/api/src/connections/api.ts b/packages/core/api/src/connections/api.ts
index 8f0430155..3bca0a444 100644
--- a/packages/core/api/src/connections/api.ts
+++ b/packages/core/api/src/connections/api.ts
@@ -3,7 +3,9 @@ import { Schema } from "effect";
 
 import {
   ConnectionId,
+  ConnectionIdentityOverride,
   ConnectionInUseError,
+  ConnectionNotFoundError,
   InternalError,
   ScopeId,
   Usage,
@@ -27,15 +29,34 @@ const ConnectionRefResponse = Schema.Struct({
   identityLabel: Schema.NullOr(Schema.String),
   expiresAt: Schema.NullOr(Schema.Number),
   oauthScope: Schema.NullOr(Schema.String),
+  identityOverride: Schema.NullOr(ConnectionIdentityOverride),
   createdAt: Schema.Number,
   updatedAt: Schema.Number,
 });
 
+export const ConnectionIdentityResponse = Schema.Struct({
+  status: Schema.Literals(["available", "unavailable", "reauth_required", "error"]),
+  source: Schema.Literals(["detected", "manual", "mixed", "unknown"]),
+  subject: Schema.NullOr(Schema.String),
+  email: Schema.NullOr(Schema.String),
+  emailVerified: Schema.NullOr(Schema.Boolean),
+  name: Schema.NullOr(Schema.String),
+  username: Schema.NullOr(Schema.String),
+  picture: Schema.NullOr(Schema.String),
+  message: Schema.NullOr(Schema.String),
+});
+export type ConnectionIdentityResponse = typeof ConnectionIdentityResponse.Type;
+
+const UpdateConnectionIdentityPayload = Schema.Struct({
+  identityOverride: Schema.NullOr(ConnectionIdentityOverride),
+});
+
 // ---------------------------------------------------------------------------
 // Group
 // ---------------------------------------------------------------------------
 
 const ConnectionInUse = ConnectionInUseError.annotate({ httpApiStatus: 409 });
+const ConnectionNotFound = ConnectionNotFoundError.annotate({ httpApiStatus: 404 });
 
 export const ConnectionsApi = HttpApiGroup.make("connections")
   .add(
@@ -58,4 +79,19 @@ export const ConnectionsApi = HttpApiGroup.make("connections")
       success: Schema.Array(Usage),
       error: InternalError,
     }),
+  )
+  .add(
+    HttpApiEndpoint.get("identity", "/scopes/:scopeId/connections/:connectionId/identity", {
+      params: ConnectionParams,
+      success: ConnectionIdentityResponse,
+      error: InternalError,
+    }),
+  )
+  .add(
+    HttpApiEndpoint.patch("updateIdentity", "/scopes/:scopeId/connections/:connectionId/identity", {
+      params: ConnectionParams,
+      payload: UpdateConnectionIdentityPayload,
+      success: ConnectionRefResponse,
+      error: [InternalError, ConnectionNotFound],
+    }),
   );
diff --git a/packages/core/api/src/handlers/connection-identity.test.ts b/packages/core/api/src/handlers/connection-identity.test.ts
new file mode 100644
index 000000000..a1a7f26c5
--- /dev/null
+++ b/packages/core/api/src/handlers/connection-identity.test.ts
@@ -0,0 +1,348 @@
+import { describe, expect, it } from "@effect/vitest";
+import { Effect, Ref } from "effect";
+import { HttpServerResponse } from "effect/unstable/http";
+
+import {
+  CreateConnectionInput,
+  OAUTH2_PROVIDER_KEY,
+  TokenMaterial,
+  createExecutor,
+} from "@executor-js/sdk";
+import { ConnectionId, ScopeId, SecretId } from "@executor-js/sdk/shared";
+import { makeTestConfig, memorySecretsPlugin, serveTestHttpApp } from "@executor-js/sdk/testing";
+
+import { lookupOidcConnectionIdentity, readConnectionIdentity } from "./connection-identity";
+
+type CapturedRequest = {
+  readonly method: string;
+  readonly url: string;
+  readonly headers: Readonly<Record<string, string>>;
+};
+
+type Handler = (request: CapturedRequest, baseUrl: string) => HttpServerResponse.HttpServerResponse;
+
+const notFound = (): HttpServerResponse.HttpServerResponse =>
+  HttpServerResponse.empty({ status: 404 });
+
+const serveOidcFixture = (handler: Handler) =>
+  Effect.gen(function* () {
+    const requests = yield* Ref.make<readonly CapturedRequest[]>([]);
+    const baseUrlRef = { value: "" };
+    const server = yield* serveTestHttpApp((request) =>
+      Effect.gen(function* () {
+        const captured = {
+          method: request.method,
+          url: request.url ?? "/",
+          headers: request.headers,
+        };
+        yield* Ref.update(requests, (all) => [...all, captured]);
+        return handler(captured, baseUrlRef.value);
+      }),
+    );
+    baseUrlRef.value = server.baseUrl;
+
+    return {
+      baseUrl: server.baseUrl,
+      requests: Ref.get(requests),
+    } as const;
+  });
+
+const withOidcFixture = <A, E>(
+  handler: Handler,
+  use: (fixture: {
+    readonly baseUrl: string;
+    readonly requests: Effect.Effect<readonly CapturedRequest[]>;
+  }) => Effect.Effect<A, E>,
+) =>
+  Effect.scoped(
+    Effect.gen(function* () {
+      const fixture = yield* serveOidcFixture(handler);
+      return yield* use(fixture);
+    }),
+  );
+
+describe("lookupOidcConnectionIdentity", () => {
+  it.effect("reads normalized account claims from OIDC userinfo", () =>
+    withOidcFixture(
+      (request, baseUrl) => {
+        if (request.url === "/.well-known/openid-configuration") {
+          return HttpServerResponse.jsonUnsafe({
+            issuer: baseUrl,
+            userinfo_endpoint: `${baseUrl}/userinfo`,
+          });
+        }
+        if (request.url === "/userinfo") {
+          return HttpServerResponse.jsonUnsafe({
+            sub: "account-123",
+            email: "rhys@example.com",
+            email_verified: true,
+            name: "Rhys Sullivan",
+            preferred_username: "rhys",
+            picture: "https://example.com/avatar.png",
+          });
+        }
+        return notFound();
+      },
+      ({ baseUrl, requests }) =>
+        Effect.gen(function* () {
+          const identity = yield* lookupOidcConnectionIdentity({
+            issuerUrl: baseUrl,
+            accessToken: "token-abc",
+          });
+          const seenRequests = yield* requests;
+
+          expect(identity).toEqual({
+            status: "available",
+            source: "detected",
+            subject: "account-123",
+            email: "rhys@example.com",
+            emailVerified: true,
+            name: "Rhys Sullivan",
+            username: "rhys",
+            picture: "https://example.com/avatar.png",
+            message: null,
+          });
+          expect(seenRequests.map((request) => request.url)).toEqual([
+            "/.well-known/openid-configuration",
+            "/userinfo",
+          ]);
+          expect(seenRequests[1]?.headers.authorization).toBe("Bearer token-abc");
+        }),
+    ),
+  );
+
+  it.effect("returns unavailable when OIDC metadata has no userinfo endpoint", () =>
+    withOidcFixture(
+      (request, baseUrl) => {
+        if (request.url === "/.well-known/openid-configuration") {
+          return HttpServerResponse.jsonUnsafe({ issuer: baseUrl });
+        }
+        return notFound();
+      },
+      ({ baseUrl, requests }) =>
+        Effect.gen(function* () {
+          const identity = yield* lookupOidcConnectionIdentity({
+            issuerUrl: baseUrl,
+            accessToken: "token-abc",
+          });
+          const seenRequests = yield* requests;
+
+          expect(identity).toEqual({
+            status: "unavailable",
+            source: "unknown",
+            subject: null,
+            email: null,
+            emailVerified: null,
+            name: null,
+            username: null,
+            picture: null,
+            message: "This connection does not advertise OIDC userinfo",
+          });
+          expect(seenRequests.map((request) => request.url)).toEqual([
+            "/.well-known/openid-configuration",
+          ]);
+        }),
+    ),
+  );
+
+  it.effect("marks the connection as needing reauth when userinfo rejects the token", () =>
+    withOidcFixture(
+      (request, baseUrl) => {
+        if (request.url === "/.well-known/openid-configuration") {
+          return HttpServerResponse.jsonUnsafe({
+            issuer: baseUrl,
+            userinfo_endpoint: `${baseUrl}/userinfo`,
+          });
+        }
+        if (request.url === "/userinfo") {
+          return HttpServerResponse.jsonUnsafe({ error: "invalid_token" }, { status: 401 });
+        }
+        return notFound();
+      },
+      ({ baseUrl }) =>
+        Effect.gen(function* () {
+          const identity = yield* lookupOidcConnectionIdentity({
+            issuerUrl: baseUrl,
+            accessToken: "expired-token",
+          });
+
+          expect(identity).toEqual({
+            status: "reauth_required",
+            source: "unknown",
+            subject: null,
+            email: null,
+            emailVerified: null,
+            name: null,
+            username: null,
+            picture: null,
+            message: "OIDC userinfo rejected the access token",
+          });
+        }),
+    ),
+  );
+
+  it.effect("returns unavailable when userinfo is outside the token's granted scopes", () =>
+    withOidcFixture(
+      (request, baseUrl) => {
+        if (request.url === "/.well-known/openid-configuration") {
+          return HttpServerResponse.jsonUnsafe({
+            issuer: baseUrl,
+            userinfo_endpoint: `${baseUrl}/userinfo`,
+          });
+        }
+        if (request.url === "/userinfo") {
+          return HttpServerResponse.jsonUnsafe({ error: "insufficient_scope" }, { status: 403 });
+        }
+        return notFound();
+      },
+      ({ baseUrl }) =>
+        Effect.gen(function* () {
+          const identity = yield* lookupOidcConnectionIdentity({
+            issuerUrl: baseUrl,
+            accessToken: "limited-token",
+          });
+
+          expect(identity).toEqual({
+            status: "unavailable",
+            source: "unknown",
+            subject: null,
+            email: null,
+            emailVerified: null,
+            name: null,
+            username: null,
+            picture: null,
+            message: "OIDC userinfo is not permitted by this token",
+          });
+        }),
+    ),
+  );
+});
+
+describe("readConnectionIdentity", () => {
+  it.effect("does not call OIDC userinfo for OAuth connections without identity scopes", () =>
+    Effect.gen(function* () {
+      const userScope = ScopeId.make("test-scope");
+      const connectionId = ConnectionId.make("gmail");
+      const executor = yield* createExecutor(
+        makeTestConfig({
+          plugins: [memorySecretsPlugin()] as const,
+        }),
+      );
+
+      yield* executor.connections.create(
+        CreateConnectionInput.make({
+          id: connectionId,
+          scope: userScope,
+          provider: OAUTH2_PROVIDER_KEY,
+          identityLabel: "Gmail API OAuth",
+          accessToken: TokenMaterial.make({
+            secretId: SecretId.make("gmail.access-token"),
+            name: "Gmail access token",
+            value: "gmail-token",
+          }),
+          refreshToken: null,
+          expiresAt: null,
+          oauthScope: "https://www.googleapis.com/auth/gmail.readonly",
+          providerState: {
+            kind: "authorization-code",
+            tokenEndpoint: "https://oauth2.googleapis.com/token",
+            issuerUrl: "https://accounts.google.com",
+            clientIdSecretId: "client-id",
+            clientIdSecretScopeId: null,
+            clientSecretSecretId: null,
+            clientSecretSecretScopeId: null,
+            clientAuth: "body",
+            scopes: ["https://www.googleapis.com/auth/gmail.readonly"],
+            scope: "https://www.googleapis.com/auth/gmail.readonly",
+          },
+        }),
+      );
+
+      const identity = yield* readConnectionIdentity({
+        executor,
+        scopeId: userScope,
+        connectionId,
+      });
+
+      expect(identity).toEqual({
+        status: "unavailable",
+        source: "unknown",
+        subject: null,
+        email: null,
+        emailVerified: null,
+        name: null,
+        username: null,
+        picture: null,
+        message: "Connection was not granted OIDC identity scopes",
+      });
+    }),
+  );
+
+  it.effect("uses manual account info when OIDC identity is unavailable", () =>
+    Effect.gen(function* () {
+      const userScope = ScopeId.make("test-scope");
+      const connectionId = ConnectionId.make("gmail");
+      const executor = yield* createExecutor(
+        makeTestConfig({
+          plugins: [memorySecretsPlugin()] as const,
+        }),
+      );
+
+      yield* executor.connections.create(
+        CreateConnectionInput.make({
+          id: connectionId,
+          scope: userScope,
+          provider: OAUTH2_PROVIDER_KEY,
+          identityLabel: "Gmail API OAuth",
+          accessToken: TokenMaterial.make({
+            secretId: SecretId.make("manual.access-token"),
+            name: "Manual access token",
+            value: "manual-token",
+          }),
+          refreshToken: null,
+          expiresAt: null,
+          oauthScope: "https://www.googleapis.com/auth/gmail.readonly",
+          providerState: {
+            kind: "authorization-code",
+            tokenEndpoint: "https://oauth2.googleapis.com/token",
+            issuerUrl: "https://accounts.google.com",
+            clientIdSecretId: "client-id",
+            clientIdSecretScopeId: null,
+            clientSecretSecretId: null,
+            clientSecretSecretScopeId: null,
+            clientAuth: "body",
+            scopes: ["https://www.googleapis.com/auth/gmail.readonly"],
+            scope: "https://www.googleapis.com/auth/gmail.readonly",
+          },
+        }),
+      );
+      yield* executor.connections.setIdentityOverride({
+        id: connectionId,
+        targetScope: userScope,
+        identityOverride: {
+          displayName: "Manual Account",
+          email: "manual@example.com",
+          avatarUrl: "https://example.com/manual.png",
+        },
+      });
+
+      const identity = yield* readConnectionIdentity({
+        executor,
+        scopeId: userScope,
+        connectionId,
+      });
+
+      expect(identity).toEqual({
+        status: "available",
+        source: "manual",
+        subject: null,
+        email: "manual@example.com",
+        emailVerified: null,
+        name: "Manual Account",
+        username: null,
+        picture: "https://example.com/manual.png",
+        message: null,
+      });
+    }),
+  );
+});
diff --git a/packages/core/api/src/handlers/connection-identity.ts b/packages/core/api/src/handlers/connection-identity.ts
new file mode 100644
index 000000000..bef8c5795
--- /dev/null
+++ b/packages/core/api/src/handlers/connection-identity.ts
@@ -0,0 +1,317 @@
+import { Data, Duration, Effect, Exit, Option, Predicate, Schema, type Layer } from "effect";
+import { FetchHttpClient, HttpClient, HttpClientRequest } from "effect/unstable/http";
+
+import {
+  OAUTH2_DEFAULT_TIMEOUT_MS,
+  OAUTH2_PROVIDER_KEY,
+  OAuthProviderStateSchema,
+  assertSupportedOAuthEndpointUrl,
+  type Executor,
+} from "@executor-js/sdk";
+import type { ConnectionId, ScopeId } from "@executor-js/sdk/shared";
+
+import type { ConnectionIdentityResponse } from "../connections/api";
+
+const OidcDiscoveryMetadata = Schema.Struct({
+  issuer: Schema.optional(Schema.String),
+  userinfo_endpoint: Schema.optional(Schema.String),
+}).annotate({ identifier: "OidcDiscoveryMetadata" });
+
+const UserInfoClaims = Schema.Struct({
+  sub: Schema.optional(Schema.String),
+  email: Schema.optional(Schema.String),
+  email_verified: Schema.optional(Schema.Boolean),
+  name: Schema.optional(Schema.String),
+  preferred_username: Schema.optional(Schema.String),
+  picture: Schema.optional(Schema.String),
+}).annotate({ identifier: "OidcUserInfoClaims" });
+
+const decodeProviderStateOption = Schema.decodeUnknownOption(OAuthProviderStateSchema);
+const decodeDiscoveryMetadataJson = Schema.decodeUnknownEffect(
+  Schema.fromJsonString(OidcDiscoveryMetadata),
+);
+const decodeUserInfoJson = Schema.decodeUnknownEffect(Schema.fromJsonString(UserInfoClaims));
+
+class ConnectionIdentityLookupError extends Data.TaggedError("ConnectionIdentityLookupError")<{
+  readonly message: string;
+  readonly status?: number;
+  readonly cause?: unknown;
+}> {}
+
+const emptyIdentity = (
+  status: ConnectionIdentityResponse["status"],
+  message: string | null,
+): ConnectionIdentityResponse => ({
+  status,
+  source: "unknown",
+  subject: null,
+  email: null,
+  emailVerified: null,
+  name: null,
+  username: null,
+  picture: null,
+  message,
+});
+
+const clean = (value: string | undefined): string | null => {
+  const trimmed = value?.trim();
+  return trimmed && trimmed.length > 0 ? trimmed : null;
+};
+
+const hasOidcIdentityScope = (oauthScope: string | null): boolean =>
+  oauthScope
+    ?.split(/\s+/)
+    .some((scope) => scope === "openid" || scope === "profile" || scope === "email") ?? false;
+
+const applyIdentityOverride = (
+  identity: ConnectionIdentityResponse,
+  override: {
+    readonly displayName: string | null;
+    readonly email: string | null;
+    readonly avatarUrl: string | null;
+  } | null,
+): ConnectionIdentityResponse => {
+  if (!override) return identity;
+  const name = clean(override.displayName ?? undefined);
+  const email = clean(override.email ?? undefined);
+  const picture = clean(override.avatarUrl ?? undefined);
+  if (!name && !email && !picture) return identity;
+  const hasDetected =
+    identity.status === "available" &&
+    Boolean(identity.name || identity.email || identity.picture || identity.subject);
+  return {
+    ...identity,
+    status: "available",
+    source: hasDetected ? "mixed" : "manual",
+    name: name ?? identity.name,
+    email: email ?? identity.email,
+    picture: picture ?? identity.picture,
+    message: hasDetected ? identity.message : null,
+  };
+};
+
+const oidcMetadataUrlFor = (issuer: string): Effect.Effect<string, ConnectionIdentityLookupError> =>
+  Effect.try({
+    try: () => {
+      assertSupportedOAuthEndpointUrl(issuer, "OIDC issuer URL");
+      const issuerUrl = new URL(issuer);
+      const issuerOrigin = `${issuerUrl.protocol}//${issuerUrl.host}`;
+      const issuerPath = issuerUrl.pathname.replace(/\/+$/, "");
+      const metadataUrl =
+        issuerPath && issuerPath !== "/"
+          ? `${issuerOrigin}/.well-known/openid-configuration${issuerPath}`
+          : `${issuerOrigin}/.well-known/openid-configuration`;
+      assertSupportedOAuthEndpointUrl(metadataUrl, "OIDC metadata URL");
+      return metadataUrl;
+    },
+    catch: (cause) =>
+      new ConnectionIdentityLookupError({
+        message: "OIDC issuer URL is not supported",
+        cause,
+      }),
+  });
+
+const executeText = (
+  request: HttpClientRequest.HttpClientRequest,
+  options: {
+    readonly httpClientLayer?: Layer.Layer<HttpClient.HttpClient>;
+    readonly timeoutMs?: number;
+  },
+  message: string,
+): Effect.Effect<
+  { readonly status: number; readonly body: string },
+  ConnectionIdentityLookupError
+> =>
+  Effect.gen(function* () {
+    const client = yield* HttpClient.HttpClient;
+    const response = yield* client.execute(request).pipe(
+      Effect.timeoutOrElse({
+        duration: Duration.millis(options.timeoutMs ?? OAUTH2_DEFAULT_TIMEOUT_MS),
+        orElse: () =>
+          Effect.fail(
+            new ConnectionIdentityLookupError({
+              message,
+              cause: "timeout",
+            }),
+          ),
+      }),
+      Effect.mapError((cause) =>
+        Predicate.isTagged("ConnectionIdentityLookupError")(cause)
+          ? cause
+          : new ConnectionIdentityLookupError({ message, cause }),
+      ),
+    );
+    const body = yield* response.text.pipe(
+      Effect.mapError(
+        (cause) =>
+          new ConnectionIdentityLookupError({
+            message: `${message}: response body could not be read`,
+            status: response.status,
+            cause,
+          }),
+      ),
+    );
+    return { status: response.status, body };
+  }).pipe(Effect.provide(options.httpClientLayer ?? FetchHttpClient.layer));
+
+const fetchOidcMetadata = (
+  issuer: string,
+  options: {
+    readonly httpClientLayer?: Layer.Layer<HttpClient.HttpClient>;
+    readonly timeoutMs?: number;
+  },
+): Effect.Effect<typeof OidcDiscoveryMetadata.Type, ConnectionIdentityLookupError> =>
+  Effect.gen(function* () {
+    const metadataUrl = yield* oidcMetadataUrlFor(issuer);
+    const response = yield* executeText(
+      HttpClientRequest.get(metadataUrl).pipe(
+        HttpClientRequest.setHeader("accept", "application/json"),
+      ),
+      options,
+      "Failed to fetch OIDC metadata",
+    );
+    if (response.status < 200 || response.status >= 300) {
+      return yield* new ConnectionIdentityLookupError({
+        message: `OIDC metadata returned status ${response.status}`,
+        status: response.status,
+      });
+    }
+    return yield* decodeDiscoveryMetadataJson(response.body).pipe(
+      Effect.mapError(
+        (cause) =>
+          new ConnectionIdentityLookupError({
+            message: "OIDC metadata is malformed",
+            cause,
+          }),
+      ),
+    );
+  });
+
+export const lookupOidcConnectionIdentity = (
+  input: {
+    readonly issuerUrl: string;
+    readonly accessToken: string;
+  },
+  options: {
+    readonly httpClientLayer?: Layer.Layer<HttpClient.HttpClient>;
+    readonly timeoutMs?: number;
+  } = {},
+): Effect.Effect<ConnectionIdentityResponse, never> =>
+  Effect.gen(function* () {
+    const metadata = yield* fetchOidcMetadata(input.issuerUrl, options).pipe(
+      Effect.catchTag("ConnectionIdentityLookupError", () => Effect.succeed(null)),
+    );
+    const advertisedUserinfoEndpoint = metadata?.userinfo_endpoint;
+    if (!advertisedUserinfoEndpoint) {
+      return emptyIdentity("unavailable", "This connection does not advertise OIDC userinfo");
+    }
+
+    const userinfoEndpoint = yield* Effect.try({
+      try: () => {
+        assertSupportedOAuthEndpointUrl(advertisedUserinfoEndpoint, "OIDC userinfo URL");
+        return advertisedUserinfoEndpoint;
+      },
+      catch: (cause) =>
+        new ConnectionIdentityLookupError({
+          message: "OIDC userinfo URL is not supported",
+          cause,
+        }),
+    }).pipe(Effect.catchTag("ConnectionIdentityLookupError", () => Effect.succeed(null)));
+    if (!userinfoEndpoint) return emptyIdentity("unavailable", "OIDC userinfo is unavailable");
+
+    const response = yield* executeText(
+      HttpClientRequest.get(userinfoEndpoint).pipe(
+        HttpClientRequest.setHeader("accept", "application/json"),
+        HttpClientRequest.setHeader("authorization", `Bearer ${input.accessToken}`),
+      ),
+      options,
+      "Failed to fetch OIDC userinfo",
+    ).pipe(
+      Effect.catchTag("ConnectionIdentityLookupError", ({ message }) =>
+        Effect.succeed({ status: 0, body: "", message } as const),
+      ),
+    );
+    if ("message" in response) return emptyIdentity("error", response.message);
+    if (response.status === 401) {
+      return emptyIdentity("reauth_required", "OIDC userinfo rejected the access token");
+    }
+    if (response.status === 403) {
+      return emptyIdentity("unavailable", "OIDC userinfo is not permitted by this token");
+    }
+    if (response.status < 200 || response.status >= 300) {
+      return emptyIdentity("error", `OIDC userinfo returned status ${response.status}`);
+    }
+
+    const claims = yield* decodeUserInfoJson(response.body).pipe(
+      Effect.catch(() => Effect.succeed(null as typeof UserInfoClaims.Type | null)),
+    );
+    if (!claims) return emptyIdentity("error", "OIDC userinfo response is malformed");
+
+    return {
+      status: "available",
+      source: "detected",
+      subject: clean(claims.sub),
+      email: clean(claims.email),
+      emailVerified: claims.email_verified ?? null,
+      name: clean(claims.name),
+      username: clean(claims.preferred_username),
+      picture: clean(claims.picture),
+      message: null,
+    };
+  });
+
+export const readConnectionIdentity = (input: {
+  readonly executor: Executor;
+  readonly scopeId: ScopeId;
+  readonly connectionId: ConnectionId;
+}): Effect.Effect<ConnectionIdentityResponse, never> =>
+  Effect.gen(function* () {
+    const connectionExit = yield* Effect.exit(
+      input.executor.connections.getAtScope(input.connectionId, input.scopeId),
+    );
+    if (Exit.isFailure(connectionExit)) {
+      return emptyIdentity("error", "Could not read connection metadata");
+    }
+    const connection = connectionExit.value;
+    if (!connection) return emptyIdentity("unavailable", "Connection was not found");
+    const withOverride = (identity: ConnectionIdentityResponse) =>
+      applyIdentityOverride(identity, connection.identityOverride);
+    if (connection.provider !== OAUTH2_PROVIDER_KEY) {
+      return withOverride(
+        emptyIdentity("unavailable", "Only OAuth2 connections can expose account identity"),
+      );
+    }
+
+    const providerState = Option.getOrNull(decodeProviderStateOption(connection.providerState));
+    const issuerUrl =
+      providerState && providerState.kind !== "client-credentials"
+        ? (providerState.issuerUrl ?? null)
+        : null;
+    if (!issuerUrl) {
+      return withOverride(
+        emptyIdentity("unavailable", "Connection does not include an OIDC issuer"),
+      );
+    }
+    if (!hasOidcIdentityScope(connection.oauthScope)) {
+      return withOverride(
+        emptyIdentity("unavailable", "Connection was not granted OIDC identity scopes"),
+      );
+    }
+
+    const accessTokenExit = yield* Effect.exit(
+      input.executor.connections.accessTokenAtScope(input.connectionId, input.scopeId),
+    );
+    if (Exit.isFailure(accessTokenExit)) {
+      const error = Option.getOrNull(Exit.findErrorOption(accessTokenExit));
+      if (error && Predicate.isTagged("ConnectionReauthRequiredError")(error)) {
+        return withOverride(emptyIdentity("reauth_required", "Connection needs re-authentication"));
+      }
+      return withOverride(emptyIdentity("error", "Could not read the connection access token"));
+    }
+
+    const identity = yield* lookupOidcConnectionIdentity({
+      issuerUrl,
+      accessToken: accessTokenExit.value,
+    });
+    return withOverride(identity);
+  });
diff --git a/packages/core/api/src/handlers/connections.ts b/packages/core/api/src/handlers/connections.ts
index 8195b16dc..24c7d9c5a 100644
--- a/packages/core/api/src/handlers/connections.ts
+++ b/packages/core/api/src/handlers/connections.ts
@@ -2,10 +2,15 @@ import { HttpApiBuilder } from "effect/unstable/httpapi";
 import { Effect } from "effect";
 
 import { capture } from "@executor-js/api";
-import { RemoveConnectionInput, type ConnectionRef } from "@executor-js/sdk";
+import {
+  RemoveConnectionInput,
+  UpdateConnectionIdentityInput,
+  type ConnectionRef,
+} from "@executor-js/sdk";
 
 import { ExecutorApi } from "../api";
 import { ExecutorService } from "../services";
+import { readConnectionIdentity } from "./connection-identity";
 
 const refToResponse = (ref: ConnectionRef) => ({
   id: ref.id,
@@ -14,6 +19,7 @@ const refToResponse = (ref: ConnectionRef) => ({
   identityLabel: ref.identityLabel,
   expiresAt: ref.expiresAt,
   oauthScope: ref.oauthScope,
+  identityOverride: ref.identityOverride,
   createdAt: ref.createdAt.getTime(),
   updatedAt: ref.updatedAt.getTime(),
 });
@@ -50,5 +56,32 @@ export const ConnectionsHandlers = HttpApiBuilder.group(ExecutorApi, "connection
           return yield* executor.connections.usages(path.connectionId);
         }),
       ),
+    )
+    .handle("identity", ({ params: path }) =>
+      capture(
+        Effect.gen(function* () {
+          const executor = yield* ExecutorService;
+          return yield* readConnectionIdentity({
+            executor,
+            scopeId: path.scopeId,
+            connectionId: path.connectionId,
+          });
+        }),
+      ),
+    )
+    .handle("updateIdentity", ({ params: path, payload }) =>
+      capture(
+        Effect.gen(function* () {
+          const executor = yield* ExecutorService;
+          const ref = yield* executor.connections.setIdentityOverride(
+            UpdateConnectionIdentityInput.make({
+              id: path.connectionId,
+              targetScope: path.scopeId,
+              identityOverride: payload.identityOverride,
+            }),
+          );
+          return refToResponse(ref);
+        }),
+      ),
     ),
 );
diff --git a/packages/core/api/src/handlers/sources.ts b/packages/core/api/src/handlers/sources.ts
index a72b2eb9c..e2f781da8 100644
--- a/packages/core/api/src/handlers/sources.ts
+++ b/packages/core/api/src/handlers/sources.ts
@@ -23,6 +23,7 @@ export const SourcesHandlers = HttpApiBuilder.group(ExecutorApi, "sources", (han
             canRemove: s.canRemove,
             canRefresh: s.canRefresh,
             canEdit: s.canEdit,
+            connectionIds: s.connectionIds,
           }));
         }),
       ),
diff --git a/packages/core/api/src/sources/api.ts b/packages/core/api/src/sources/api.ts
index e61889279..205ed6dfe 100644
--- a/packages/core/api/src/sources/api.ts
+++ b/packages/core/api/src/sources/api.ts
@@ -37,6 +37,7 @@ const SourceResponse = Schema.Struct({
   canRemove: Schema.optional(Schema.Boolean),
   canRefresh: Schema.optional(Schema.Boolean),
   canEdit: Schema.optional(Schema.Boolean),
+  connectionIds: Schema.optional(Schema.Array(Schema.String)),
 });
 
 const SourceRemoveResponse = Schema.Struct({
diff --git a/packages/core/execution/src/promise.ts b/packages/core/execution/src/promise.ts
index 6333746b8..e49d25aa7 100644
--- a/packages/core/execution/src/promise.ts
+++ b/packages/core/execution/src/promise.ts
@@ -117,6 +117,7 @@ const wrapPromiseExecutor = (pe: PromiseExecutor): EffectExecutor => ({
     create: (input) => fromPromise(() => pe.connections.create(input)),
     updateTokens: (input) => fromPromise(() => pe.connections.updateTokens(input)),
     setIdentityLabel: (id, label) => fromPromise(() => pe.connections.setIdentityLabel(id, label)),
+    setIdentityOverride: (input) => fromPromise(() => pe.connections.setIdentityOverride(input)),
     accessToken: (id) => fromPromise(() => pe.connections.accessToken(id)),
     accessTokenAtScope: (id, scope) =>
       fromPromise(() => pe.connections.accessTokenAtScope(id, scope)),
diff --git a/packages/core/sdk/src/connections.ts b/packages/core/sdk/src/connections.ts
index ade873e5c..cb2c6d42f 100644
--- a/packages/core/sdk/src/connections.ts
+++ b/packages/core/sdk/src/connections.ts
@@ -18,6 +18,13 @@ import { ConnectionId, ScopeId, SecretId } from "./ids";
 export const ConnectionProviderState = Schema.Record(Schema.String, Schema.Unknown);
 export type ConnectionProviderState = typeof ConnectionProviderState.Type;
 
+export const ConnectionIdentityOverride = Schema.Struct({
+  displayName: Schema.NullOr(Schema.String),
+  email: Schema.NullOr(Schema.String),
+  avatarUrl: Schema.NullOr(Schema.String),
+});
+export type ConnectionIdentityOverride = typeof ConnectionIdentityOverride.Type;
+
 // ---------------------------------------------------------------------------
 // ConnectionRef — metadata projection returned from `ctx.connections.list`
 // / `executor.connections.list`. Holds token secret ids (so a plugin can
@@ -37,6 +44,7 @@ export const ConnectionRef = Schema.Struct({
    *  `oauthScope` to avoid collision with the executor scope id. */
   oauthScope: Schema.NullOr(Schema.String),
   providerState: Schema.NullOr(ConnectionProviderState),
+  identityOverride: Schema.NullOr(ConnectionIdentityOverride),
   createdAt: Schema.Date,
   updatedAt: Schema.Date,
 });
@@ -178,6 +186,13 @@ export const UpdateConnectionTokensInput = Schema.Struct({
 });
 export type UpdateConnectionTokensInput = typeof UpdateConnectionTokensInput.Type;
 
+export const UpdateConnectionIdentityInput = Schema.Struct({
+  id: ConnectionId,
+  targetScope: ScopeId,
+  identityOverride: Schema.NullOr(ConnectionIdentityOverride),
+});
+export type UpdateConnectionIdentityInput = typeof UpdateConnectionIdentityInput.Type;
+
 export const RemoveConnectionInput = Schema.Struct({
   id: ConnectionId,
   /** Scope id whose connection row and owned token secrets should be removed. */
diff --git a/packages/core/sdk/src/core-schema.ts b/packages/core/sdk/src/core-schema.ts
index 491cb190e..f3808cd7e 100644
--- a/packages/core/sdk/src/core-schema.ts
+++ b/packages/core/sdk/src/core-schema.ts
@@ -180,6 +180,7 @@ export const coreTables = defineTables({
     expires_at: nullableBigintColumn("expires_at"),
     scope: nullableTextColumn("scope"),
     provider_state: nullableJsonColumn("provider_state"),
+    identity_override: nullableJsonColumn("identity_override"),
     created_at: dateColumn("created_at"),
     updated_at: dateColumn("updated_at"),
   }),
diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts
index 0dcd20563..35d858347 100644
--- a/packages/core/sdk/src/executor.ts
+++ b/packages/core/sdk/src/executor.ts
@@ -32,11 +32,13 @@ import { makeFumaBlobStore, pluginBlobStore } from "./blob";
 import { coreToolsPlugin } from "./core-tools";
 import {
   ConnectionProviderState,
+  ConnectionIdentityOverride,
   ConnectionRef,
   ConnectionRefreshError,
   type ConnectionProvider,
   type ConnectionRefreshResult,
   type CreateConnectionInput,
+  type UpdateConnectionIdentityInput,
   type RemoveConnectionInput,
   type UpdateConnectionTokensInput,
 } from "./connections";
@@ -317,6 +319,9 @@ export type Executor<TPlugins extends readonly AnyPlugin[] = readonly []> = {
       id: string,
       label: string | null,
     ) => Effect.Effect<void, ConnectionNotFoundError | StorageFailure>;
+    readonly setIdentityOverride: (
+      input: UpdateConnectionIdentityInput,
+    ) => Effect.Effect<ConnectionRef, ConnectionNotFoundError | StorageFailure>;
     readonly accessToken: (
       id: string,
     ) => Effect.Effect<
@@ -495,7 +500,7 @@ const createDefaultMemoryDb = (tables: FumaTables): ExecutorDb => {
 // Row → public projection conversions
 // ---------------------------------------------------------------------------
 
-const rowToSource = (row: SourceRow): Source => ({
+const rowToSource = (row: SourceRow, connectionIds: readonly string[] = []): Source => ({
   id: row.id,
   scopeId: row.scope_id,
   kind: row.kind,
@@ -505,6 +510,7 @@ const rowToSource = (row: SourceRow): Source => ({
   canRemove: Boolean(row.can_remove),
   canRefresh: Boolean(row.can_refresh),
   canEdit: Boolean(row.can_edit),
+  connectionIds,
   runtime: false,
 });
 
@@ -518,6 +524,7 @@ const staticDeclToSource = (decl: StaticSourceDecl, pluginId: string): Source =>
   canRemove: decl.canRemove ?? false,
   canRefresh: decl.canRefresh ?? false,
   canEdit: decl.canEdit ?? false,
+  connectionIds: [],
   runtime: true,
 });
 
@@ -530,6 +537,7 @@ const decodeJsonColumn = (value: unknown): unknown => {
 };
 
 const decodeProviderState = Schema.decodeUnknownOption(ConnectionProviderState);
+const decodeConnectionIdentityOverride = Schema.decodeUnknownOption(ConnectionIdentityOverride);
 
 const rowToTool = (row: ToolRow, annotations?: ToolAnnotations): Tool => ({
   id: row.id,
@@ -1998,6 +2006,9 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
         expiresAt: row.expires_at != null ? Number(row.expires_at) : null,
         oauthScope: row.scope ?? null,
         providerState: Option.getOrNull(decodeProviderState(decodeJsonColumn(row.provider_state))),
+        identityOverride: Option.getOrNull(
+          decodeConnectionIdentityOverride(decodeJsonColumn(row.identity_override)),
+        ),
         createdAt: row.created_at instanceof Date ? row.created_at : new Date(row.created_at),
         updatedAt: row.updated_at instanceof Date ? row.updated_at : new Date(row.updated_at),
       });
@@ -2173,6 +2184,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
               expires_at: input.expiresAt ?? null,
               scope: input.oauthScope ?? null,
               provider_state: input.providerState ?? null,
+              identity_override: null,
               created_at: now,
               updated_at: now,
             });
@@ -2187,6 +2199,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
               expiresAt: input.expiresAt,
               oauthScope: input.oauthScope,
               providerState: input.providerState,
+              identityOverride: null,
               createdAt: now,
               updatedAt: now,
             });
@@ -2286,6 +2299,39 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
         });
       });
 
+    const connectionsSetIdentityOverride = (
+      input: UpdateConnectionIdentityInput,
+    ): Effect.Effect<ConnectionRef, ConnectionNotFoundError | StorageFailure> =>
+      Effect.gen(function* () {
+        yield* assertScopeInStack("connection identity targetScope", input.targetScope);
+        const row = yield* findConnectionRowAtScope({
+          connectionId: input.id,
+          scopeId: input.targetScope,
+        });
+        if (!row) {
+          return yield* new ConnectionNotFoundError({
+            connectionId: input.id,
+          });
+        }
+        yield* core.updateMany("connection", {
+          where: byScopedId(input.targetScope, input.id),
+          set: {
+            identity_override: input.identityOverride,
+            updated_at: new Date(),
+          },
+        });
+        const updated = yield* findConnectionRowAtScope({
+          connectionId: input.id,
+          scopeId: input.targetScope,
+        });
+        if (!updated) {
+          return yield* new ConnectionNotFoundError({
+            connectionId: input.id,
+          });
+        }
+        return rowToConnection(updated);
+      });
+
     const connectionsRemove = (
       input: RemoveConnectionInput,
     ): Effect.Effect<void, ConnectionInUseError | StorageFailure> =>
@@ -3411,6 +3457,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
           create: (input) => connectionsCreate(input),
           updateTokens: (input) => connectionsUpdateTokens(input),
           setIdentityLabel: (id, label) => connectionsSetIdentityLabel(id, label),
+          setIdentityOverride: (input) => connectionsSetIdentityOverride(input),
           accessToken: (id) => connectionsAccessToken(id),
           accessTokenAtScope: (id, scope) => connectionsAccessTokenAtScope(id, scope),
           remove: (input) => connectionsRemove(input),
@@ -3550,11 +3597,32 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
           }
         }
         const dynamicDeduped = [...byId.values()];
+        const sourceKeys = new Set(dynamicDeduped.map((row) => `${row.scope_id}\u0000${row.id}`));
+        const sourceConnectionIds = new Map<string, string[]>();
+        if (sourceKeys.size > 0) {
+          const bindingRows = yield* core.findMany("credential_binding", {
+            where: scopedWhere(scopeIds, (b) => b("kind", "=", "connection")),
+          });
+          for (const row of bindingRows as readonly CredentialBindingRow[]) {
+            if (!row.connection_id) continue;
+            const key = `${String(row.source_scope_id)}\u0000${String(row.source_id)}`;
+            if (!sourceKeys.has(key)) continue;
+            const connectionId = String(row.connection_id);
+            const values = sourceConnectionIds.get(key) ?? [];
+            if (!values.includes(connectionId)) values.push(connectionId);
+            sourceConnectionIds.set(key, values);
+          }
+        }
         const staticList: Source[] = [];
         for (const { source, pluginId } of staticSources.values()) {
           staticList.push(staticDeclToSource(source, pluginId));
         }
-        const merged = [...staticList, ...dynamicDeduped.map(rowToSource)];
+        const merged = [
+          ...staticList,
+          ...dynamicDeduped.map((row) =>
+            rowToSource(row, sourceConnectionIds.get(`${row.scope_id}\u0000${row.id}`) ?? []),
+          ),
+        ];
         yield* Effect.annotateCurrentSpan({
           "executor.sources.static_count": staticList.length,
           "executor.sources.dynamic_count": dynamicDeduped.length,
@@ -4431,6 +4499,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
         create: connectionsCreate,
         updateTokens: connectionsUpdateTokens,
         setIdentityLabel: connectionsSetIdentityLabel,
+        setIdentityOverride: connectionsSetIdentityOverride,
         accessToken: connectionsAccessToken,
         accessTokenAtScope: connectionsAccessTokenAtScope,
         remove: connectionsRemove,
diff --git a/packages/core/sdk/src/index.ts b/packages/core/sdk/src/index.ts
index 179998a91..2a2c60a2a 100644
--- a/packages/core/sdk/src/index.ts
+++ b/packages/core/sdk/src/index.ts
@@ -167,9 +167,11 @@ export { Usage, type UsagesForSecretInput, type UsagesForConnectionInput } from
 // Connections
 export {
   ConnectionRef,
+  ConnectionIdentityOverride,
   ConnectionProviderState,
   CreateConnectionInput,
   RemoveConnectionInput,
+  UpdateConnectionIdentityInput,
   UpdateConnectionTokensInput,
   TokenMaterial,
   ConnectionRefreshError,
diff --git a/packages/core/sdk/src/plugin.ts b/packages/core/sdk/src/plugin.ts
index e32213ae0..cb74e4f9b 100644
--- a/packages/core/sdk/src/plugin.ts
+++ b/packages/core/sdk/src/plugin.ts
@@ -12,6 +12,7 @@ import type {
   ConnectionRefreshError,
   CreateConnectionInput,
   RemoveConnectionInput,
+  UpdateConnectionIdentityInput,
   UpdateConnectionTokensInput,
 } from "./connections";
 import type {
@@ -229,6 +230,9 @@ export interface PluginCtx<TStore = unknown> {
       id: string,
       label: string | null,
     ) => Effect.Effect<void, ConnectionNotFoundError | StorageFailure>;
+    readonly setIdentityOverride: (
+      input: UpdateConnectionIdentityInput,
+    ) => Effect.Effect<ConnectionRef, ConnectionNotFoundError | StorageFailure>;
     /** Get a guaranteed-fresh access token. Calls the provider's
      *  `refresh` handler if `expires_at` is in the past / within the
      *  refresh skew window. */
diff --git a/packages/core/sdk/src/promise.ts b/packages/core/sdk/src/promise.ts
index 1615d4c44..67cbb9d87 100644
--- a/packages/core/sdk/src/promise.ts
+++ b/packages/core/sdk/src/promise.ts
@@ -16,7 +16,7 @@ export {
 // these to type arguments they pass in (e.g. SetSecretInput, filters).
 export { ScopeId, ToolId, SecretId, PolicyId } from "./ids";
 export { Scope } from "./scope";
-export { RemoveConnectionInput } from "./connections";
+export { RemoveConnectionInput, UpdateConnectionIdentityInput } from "./connections";
 export { RemoveSecretInput, SecretRef, SetSecretInput } from "./secrets";
 export type {
   CreateToolPolicyInput,
diff --git a/packages/core/sdk/src/shared.ts b/packages/core/sdk/src/shared.ts
index ccecebcae..275849a88 100644
--- a/packages/core/sdk/src/shared.ts
+++ b/packages/core/sdk/src/shared.ts
@@ -15,6 +15,7 @@ export {
   SecretResolutionError,
   SecretOwnedByConnectionError,
   SecretInUseError,
+  ConnectionNotFoundError,
   ConnectionInUseError,
 } from "./errors";
 
@@ -90,6 +91,8 @@ export { SourceDetectionResult, type Source } from "./types";
 
 export { Usage } from "./usages";
 
+export { ConnectionIdentityOverride, UpdateConnectionIdentityInput } from "./connections";
+
 export {
   DEFAULT_EXECUTOR_SERVER_ORIGIN,
   DEFAULT_EXECUTOR_SERVER_USERNAME,
diff --git a/packages/core/sdk/src/types.ts b/packages/core/sdk/src/types.ts
index 4fd67ad93..096e6b9c7 100644
--- a/packages/core/sdk/src/types.ts
+++ b/packages/core/sdk/src/types.ts
@@ -30,6 +30,8 @@ export interface Source {
    *  (`executor.openapi.updateSource(id, patch)` etc.) — this flag is
    *  just a UI signal. */
   readonly canEdit: boolean;
+  /** Connection ids currently bound to this source through shared credential slots. */
+  readonly connectionIds: readonly string[];
   /** True if the source was declared statically by a plugin at startup
    *  (in-memory only, no DB row). False if it was added at runtime via
    *  `ctx.core.sources.register(...)`. UI differentiates built-in vs