diff --git a/.changeset/scoped-introspection.md b/.changeset/scoped-introspection.md
new file mode 100644
index 0000000000..8c9adbc30f
--- /dev/null
+++ b/.changeset/scoped-introspection.md
@@ -0,0 +1,7 @@
+---
+"graphile-build-pg": minor
+"pg-introspection": minor
+---
+
+Add opt-in schema-scoped PostgreSQL introspection with transitive dependency
+closure and fail-closed dependency completeness validation.
diff --git a/graphile-build/graphile-build-pg/README.md b/graphile-build/graphile-build-pg/README.md
index a281135434..2856a50c04 100644
--- a/graphile-build/graphile-build-pg/README.md
+++ b/graphile-build/graphile-build-pg/README.md
@@ -16,6 +16,58 @@ creates the relevant GraphQL types, fields, and [grafast][] plan resolver
 functions. The result is a high-performance, powerful, auto-generated but highly
 flexible GraphQL schema.
 
+## Schema-scoped introspection
+
+PostgreSQL services can opt into schema-scoped introspection through gather
+options keyed by service name. Use `true` to enable it with defaults, `false` to
+explicitly disable it, or an options object to customize it. Services without an
+entry continue to use the full catalog query.
+
+```ts
+const preset = {
+  // ...
+  gather: {
+    pgScopedIntrospection: {
+      main: true,
+    },
+  },
+};
+```
+
+For advanced configuration:
+
+```ts
+const preset = {
+  pgServices: [
+    makePgService({
+      name: "main",
+      connectionString: process.env.DATABASE_URL,
+      schemas: ["app_public"],
+    }),
+  ],
+  gather: {
+    pgScopedIntrospection: {
+      main: {
+        catalogTypes: "dependency-closure" as const,
+        capabilityExtensions: ["pg_trgm"],
+      },
+    },
+  },
+};
+```
+
+The service's `schemas` are the roots of the introspection query. Referenced
+objects in other schemas are discovered and retained automatically, while
+unrelated objects are excluded. Configuration for an unknown service name fails
+rather than being silently ignored.
+
+Extensions required by retained objects, such as the operator class behind a
+`pg_trgm` index, are discovered automatically. `capabilityExtensions` is for a
+different case: it retains lightweight metadata proving that an extension is
+installed even when no retained object directly depends on it. For example, a
+plugin can check for `pg_trgm` before exposing an optional search capability. It
+does not install the extension or retain every object owned by it.
+
 If you don't want to use your database introspection results to generate the
 schema, you can instead build the registry yourself giving you full control over
 what goes into your GraphQL API whilst still saving you significant effort
diff --git a/graphile-build/graphile-build-pg/__tests__/fixtures/scoped-introspection.sql b/graphile-build/graphile-build-pg/__tests__/fixtures/scoped-introspection.sql
new file mode 100644
index 0000000000..4da2a79a9a
--- /dev/null
+++ b/graphile-build/graphile-build-pg/__tests__/fixtures/scoped-introspection.sql
@@ -0,0 +1,123 @@
+create schema scope_root;
+create schema scope_dependency;
+create schema scope_unrelated;
+create schema scope_extension;
+create schema scope_capability_root;
+
+create extension pg_trgm with schema scope_extension;
+
+create type scope_dependency.item_status as enum (
+  'draft',
+  'active',
+  'archived'
+);
+
+create domain scope_dependency.positive_integer as integer
+  check (value > 0);
+
+create type scope_dependency.item_payload as (
+  status scope_dependency.item_status,
+  score scope_dependency.positive_integer
+);
+
+create type scope_dependency.integer_span as range (
+  subtype = integer,
+  multirange_type_name = scope_dependency.integer_span_set
+);
+
+create table scope_dependency.dependency_owners (
+  id bigint generated always as identity primary key,
+  status scope_dependency.item_status not null
+);
+
+create table scope_dependency.inherited_base (
+  inherited_status scope_dependency.item_status not null
+);
+
+create table scope_root.closure_items (
+  id bigint generated always as identity primary key,
+  dependency_owner_id bigint not null
+    references scope_dependency.dependency_owners (id),
+  title text not null,
+  status scope_dependency.item_status not null,
+  score scope_dependency.positive_integer not null,
+  payload scope_dependency.item_payload not null,
+  active_span scope_dependency.integer_span
+);
+
+create table scope_root.inherited_items (
+  id bigint generated always as identity primary key
+) inherits (scope_dependency.inherited_base);
+
+create table scope_root.inheritance_root (
+  id bigint generated always as identity primary key,
+  root_note text not null
+);
+
+create table scope_dependency.reverse_inherited_item (
+  dependency_note text not null
+) inherits (scope_root.inheritance_root);
+
+create index closure_items_status_idx
+  on scope_root.closure_items (status);
+
+create index closure_items_title_gin_trgm_idx
+  on scope_root.closure_items
+  using gin (title scope_extension.gin_trgm_ops);
+
+create index closure_items_title_gist_trgm_idx
+  on scope_root.closure_items
+  using gist (title scope_extension.gist_trgm_ops(siglen = 32));
+
+create function scope_root.echo_dependency_status(
+  input_status scope_dependency.item_status
+)
+returns scope_dependency.item_status
+language sql
+immutable
+strict
+parallel safe
+as $$
+  select input_status;
+$$;
+
+create function scope_root.make_dependency_payload(
+  input_status scope_dependency.item_status,
+  input_score scope_dependency.positive_integer
+)
+returns scope_dependency.item_payload
+language sql
+immutable
+strict
+parallel safe
+as $$
+  select row(input_status, input_score)::scope_dependency.item_payload;
+$$;
+
+create type scope_unrelated.item_status as enum (
+  'draft',
+  'active',
+  'archived'
+);
+
+create table scope_unrelated.closure_items (
+  id bigint generated always as identity primary key,
+  status scope_unrelated.item_status not null
+);
+
+create function scope_unrelated.echo_dependency_status(
+  input_status scope_unrelated.item_status
+)
+returns scope_unrelated.item_status
+language sql
+immutable
+strict
+parallel safe
+as $$
+  select input_status;
+$$;
+
+create table scope_capability_root.capability_items (
+  id bigint generated always as identity primary key,
+  title text not null
+);
diff --git a/graphile-build/graphile-build-pg/__tests__/scopedIntrospection.integration.test.ts b/graphile-build/graphile-build-pg/__tests__/scopedIntrospection.integration.test.ts
new file mode 100644
index 0000000000..82344de590
--- /dev/null
+++ b/graphile-build/graphile-build-pg/__tests__/scopedIntrospection.integration.test.ts
@@ -0,0 +1,286 @@
+import { createHash } from "node:crypto";
+import { readFile } from "node:fs/promises";
+import { join } from "node:path";
+
+import { makePgService } from "@dataplan/pg/adaptors/pg";
+import {
+  execute,
+  type GraphQLSchema,
+  lexicographicSortSchema,
+  parse,
+  printSchema,
+} from "grafast/graphql";
+import {
+  defaultPreset as graphileBuildPreset,
+  makeSchema,
+} from "graphile-build";
+import type { Pool } from "pg";
+import pg from "pg";
+import type { Introspection } from "pg-introspection";
+
+import {
+  createTestDatabase,
+  dropTestDatabase,
+} from "../../../grafast/dataplan-pg/__tests__/sharedHelpers.ts";
+import { defaultPreset as graphileBuildPgPreset } from "../src/preset.ts";
+
+const ROOT_SCHEMA = "scope_root";
+const DEPENDENCY_SCHEMA = "scope_dependency";
+const UNRELATED_SCHEMA = "scope_unrelated";
+const EXTENSION_SCHEMA = "scope_extension";
+const CAPABILITY_ROOT_SCHEMA = "scope_capability_root";
+
+interface SchemaBuild {
+  schema: GraphQLSchema;
+  introspection: Introspection;
+  hash: string;
+}
+
+const makeCapturePlugin = (
+  capture: (introspection: Introspection) => void,
+): GraphileConfig.Plugin => ({
+  name: "ScopedIntrospectionCapturePlugin",
+  gather: {
+    namespace: "scopedIntrospectionCapture",
+    hooks: {
+      pgIntrospection_introspection(_info, event) {
+        capture(event.introspection);
+      },
+    },
+  },
+});
+
+const buildSchema = async (
+  pool: Pool,
+  scoped: boolean,
+  rootSchema = ROOT_SCHEMA,
+): Promise<SchemaBuild> => {
+  let introspection: Introspection | undefined;
+  const service = makePgService({
+    pool,
+    schemas: [rootSchema],
+    pubsub: false,
+  });
+
+  try {
+    const result = await makeSchema({
+      extends: [graphileBuildPreset, graphileBuildPgPreset],
+      disablePlugins: ["PgEnumTablesPlugin"],
+      ...(scoped
+        ? {
+            gather: {
+              pgScopedIntrospection: {
+                [service.name]: {
+                  catalogTypes: "dependency-closure" as const,
+                  capabilityExtensions: ["pg_trgm"],
+                },
+              },
+            },
+          }
+        : null),
+      plugins: [
+        makeCapturePlugin((value) => {
+          introspection = value;
+        }),
+      ],
+      pgServices: [service],
+    });
+    if (!introspection) {
+      throw new Error(
+        "PostgreSQL introspection lifecycle event was not emitted",
+      );
+    }
+    const sdl = printSchema(lexicographicSortSchema(result.schema));
+    return {
+      schema: result.schema,
+      introspection,
+      hash: createHash("sha256").update(sdl).digest("hex"),
+    };
+  } finally {
+    await service.release?.();
+  }
+};
+
+describe("schema-scoped PostgreSQL introspection", () => {
+  let databaseName = "";
+  let pool: Pool;
+  let stock: SchemaBuild;
+  let scoped: SchemaBuild;
+
+  beforeAll(async () => {
+    const testDatabase = await createTestDatabase();
+    databaseName = testDatabase.databaseName;
+    pool = new pg.Pool({ connectionString: testDatabase.connectionString });
+    const fixture = await readFile(
+      join(__dirname, "fixtures/scoped-introspection.sql"),
+      "utf8",
+    );
+    await pool.query(fixture);
+    stock = await buildSchema(pool, false);
+    scoped = await buildSchema(pool, true);
+  }, 120_000);
+
+  afterAll(async () => {
+    await pool?.end();
+    await dropTestDatabase(databaseName);
+  });
+
+  it("builds the same schema and a working runtime", async () => {
+    expect(scoped.hash).toBe(stock.hash);
+
+    const document = parse("{ __typename }");
+    const stockResult = await execute({ schema: stock.schema, document });
+    const scopedResult = await execute({ schema: scoped.schema, document });
+    expect(scopedResult).toEqual(stockResult);
+    expect(scopedResult.errors).toBeUndefined();
+    expect(scopedResult.data?.__typename).toBe("Query");
+  });
+
+  it("retains transitive table, function, and range type dependencies", () => {
+    const namespaceNames = scoped.introspection.namespaces.map(
+      (namespace) => namespace.nspname,
+    );
+    expect(namespaceNames).toEqual(
+      expect.arrayContaining([
+        ROOT_SCHEMA,
+        DEPENDENCY_SCHEMA,
+        EXTENSION_SCHEMA,
+        "pg_catalog",
+      ]),
+    );
+    expect(namespaceNames).not.toContain(UNRELATED_SCHEMA);
+
+    const rootTable = scoped.introspection.classes.find(
+      (entity) =>
+        entity.relname === "closure_items" &&
+        entity.getNamespace()?.nspname === ROOT_SCHEMA,
+    );
+    expect(rootTable).toBeDefined();
+    const attributeTypes = new Map(
+      rootTable!
+        .getAttributes()
+        .map((attribute) => [attribute.attname, attribute.getType()]),
+    );
+    expect(attributeTypes.get("status")?.typname).toBe("item_status");
+    expect(attributeTypes.get("score")?.typname).toBe("positive_integer");
+    expect(attributeTypes.get("payload")?.typname).toBe("item_payload");
+    expect(attributeTypes.get("active_span")?.typname).toBe("integer_span");
+
+    const statusType = attributeTypes.get("status");
+    expect(statusType?.getEnumValues().map((value) => value.enumlabel)).toEqual(
+      ["draft", "active", "archived"],
+    );
+    expect(statusType?.getArrayType()?.typname).toBe("_item_status");
+
+    const payloadType = attributeTypes.get("payload");
+    expect(
+      payloadType
+        ?.getClass()
+        ?.getAttributes()
+        .map((attribute) => attribute.getType()?.typname),
+    ).toEqual(["item_status", "positive_integer"]);
+
+    const echoStatus = scoped.introspection.procs.find(
+      (proc) =>
+        proc.proname === "echo_dependency_status" &&
+        proc.getNamespace()?.nspname === ROOT_SCHEMA,
+    );
+    expect(echoStatus?.getReturnType()?.typname).toBe("item_status");
+    expect(
+      echoStatus?.getArguments().map((argument) => argument.type.typname),
+    ).toEqual(["item_status"]);
+
+    const makePayload = scoped.introspection.procs.find(
+      (proc) =>
+        proc.proname === "make_dependency_payload" &&
+        proc.getNamespace()?.nspname === ROOT_SCHEMA,
+    );
+    expect(makePayload?.getReturnType()?.typname).toBe("item_payload");
+    expect(
+      makePayload?.getArguments().map((argument) => argument.type.typname),
+    ).toEqual(["item_status", "positive_integer"]);
+
+    const range = scoped.introspection.ranges.find(
+      (entity) => entity.getType()?.typname === "integer_span",
+    );
+    expect(range?.getSubType()?.typname).toBe("int4");
+    expect(
+      scoped.introspection.types.find(
+        (type) => type._id === range?.rngmultitypid,
+      )?.typname,
+    ).toBe("integer_span_set");
+
+    const foreignKey = rootTable
+      ?.getConstraints()
+      .find((constraint) => constraint.contype === "f");
+    expect(foreignKey?.getForeignClass()?.relname).toBe("dependency_owners");
+    expect(foreignKey?.getForeignClass()?.getNamespace()?.nspname).toBe(
+      DEPENDENCY_SCHEMA,
+    );
+
+    const inheritedItems = scoped.introspection.classes.find(
+      (entity) =>
+        entity.relname === "inherited_items" &&
+        entity.getNamespace()?.nspname === ROOT_SCHEMA,
+    );
+    const inherited = inheritedItems?.getInherited();
+    expect(inherited).toHaveLength(1);
+    expect(
+      scoped.introspection.classes.find(
+        (entity) => entity._id === inherited?.[0]?.inhparent,
+      )?.relname,
+    ).toBe("inherited_base");
+    expect(
+      scoped.introspection.classes.some(
+        (entity) => entity.relname === "reverse_inherited_item",
+      ),
+    ).toBe(false);
+  });
+
+  it("retains indexes and identifies their owning extension", () => {
+    const indexNames = scoped.introspection.indexes.map(
+      (index) => index.getIndexClass()?.relname,
+    );
+    expect(indexNames).toEqual(
+      expect.arrayContaining([
+        "closure_items_status_idx",
+        "closure_items_title_gin_trgm_idx",
+        "closure_items_title_gist_trgm_idx",
+      ]),
+    );
+    expect(
+      scoped.introspection.extensions.some(
+        (extension) => extension.extname === "pg_trgm",
+      ),
+    ).toBe(true);
+    expect(
+      scoped.introspection.types.some(
+        (type) => type.getNamespace()?.nspname === UNRELATED_SCHEMA,
+      ),
+    ).toBe(false);
+    expect(
+      scoped.introspection.procs.some(
+        (proc) => proc.getNamespace()?.nspname === UNRELATED_SCHEMA,
+      ),
+    ).toBe(false);
+  });
+
+  it("retains explicitly requested extension capability metadata", async () => {
+    const capabilityOnly = await buildSchema(
+      pool,
+      true,
+      CAPABILITY_ROOT_SCHEMA,
+    );
+
+    expect(
+      capabilityOnly.introspection.extensions.some(
+        (extension) => extension.extname === "pg_trgm",
+      ),
+    ).toBe(true);
+    expect(
+      capabilityOnly.introspection.indexes.some((index) =>
+        index.getIndexClass()?.relname.includes("trgm"),
+      ),
+    ).toBe(false);
+  });
+});
diff --git a/graphile-build/graphile-build-pg/__tests__/scopedIntrospection.test.ts b/graphile-build/graphile-build-pg/__tests__/scopedIntrospection.test.ts
new file mode 100644
index 0000000000..47d0b53fda
--- /dev/null
+++ b/graphile-build/graphile-build-pg/__tests__/scopedIntrospection.test.ts
@@ -0,0 +1,54 @@
+import { makeIntrospectionQuery } from "pg-introspection";
+
+import {
+  assertScopedIntrospectionServices,
+  getIntrospectionQuery,
+} from "../src/scopedIntrospection.ts";
+
+const makeService = (
+  options: Partial<GraphileConfig.PgServiceConfiguration> = {},
+): GraphileConfig.PgServiceConfiguration =>
+  ({
+    name: "main",
+    schemas: ["app_public"],
+    ...options,
+  }) as GraphileConfig.PgServiceConfiguration;
+
+describe("scoped introspection service configuration", () => {
+  it.each([undefined, false])("uses stock introspection for %p", (config) => {
+    const plan = getIntrospectionQuery(makeService(), config);
+
+    expect(plan).toEqual({
+      query: { text: makeIntrospectionQuery() },
+      requiredSchemas: null,
+      catalogTypes: null,
+    });
+  });
+
+  it("uses scoped introspection defaults for true", () => {
+    const plan = getIntrospectionQuery(makeService(), true);
+
+    expect(plan.query.values).toEqual([["app_public"], []]);
+    expect(plan.requiredSchemas).toEqual(["app_public"]);
+    expect(plan.catalogTypes).toBe("all");
+  });
+
+  it("builds a scoped, parameterized query from the service schemas", () => {
+    const plan = getIntrospectionQuery(makeService(), {
+      catalogTypes: "dependency-closure",
+      capabilityExtensions: ["pg_trgm"],
+    });
+
+    expect(plan.query.values).toEqual([["app_public"], ["pg_trgm"]]);
+    expect(plan.requiredSchemas).toEqual(["app_public"]);
+    expect(plan.catalogTypes).toBe("dependency-closure");
+  });
+
+  it("rejects configuration for an unknown PostgreSQL service", () => {
+    expect(() =>
+      assertScopedIntrospectionServices([makeService()], {
+        analytics: {},
+      }),
+    ).toThrow(/unknown PostgreSQL service\(s\): analytics/);
+  });
+});
diff --git a/graphile-build/graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts b/graphile-build/graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts
index bfab4a8cc9..3229823ba9 100644
--- a/graphile-build/graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts
+++ b/graphile-build/graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts
@@ -27,11 +27,15 @@ import type {
   PgRoles,
   PgType,
 } from "pg-introspection";
-import {
-  makeIntrospectionQuery,
-  parseIntrospectionResults,
-} from "pg-introspection";
+import { parseIntrospectionResults } from "pg-introspection";
 
+import {
+  assertDependencyClosureTypes,
+  assertScopedIntrospectionServices,
+  assertScopedNamespaces,
+  getIntrospectionQuery,
+  type IntrospectionScope,
+} from "../scopedIntrospection.ts";
 import { version } from "../version.ts";
 import { watchFixtures } from "../watchFixtures.ts";
 
@@ -242,10 +246,12 @@ declare global {
   }
 }
 
-type RawIntrospectionResults = Array<{
-  pgService: GraphileConfig.PgServiceConfiguration;
-  introspectionText: string;
-}>;
+type RawIntrospectionResults = Array<
+  {
+    pgService: GraphileConfig.PgServiceConfiguration;
+    introspectionText: string;
+  } & IntrospectionScope
+>;
 type IntrospectionResults = Array<{
   pgService: GraphileConfig.PgServiceConfiguration;
   introspection: Introspection;
@@ -534,6 +540,7 @@ export const PgIntrospectionPlugin: GraphileConfig.Plugin = {
               info.cache.introspectionResultsPromise ??
               (info.cache.introspectionResultsPromise = introspectPgServices(
                 info.resolvedPreset.pgServices,
+                info.options.pgScopedIntrospection,
               ));
 
             // Don't cache errors
@@ -544,11 +551,27 @@ export const PgIntrospectionPlugin: GraphileConfig.Plugin = {
             const rawIntrospections = await introspectionPromise;
 
             const introspections: IntrospectionResults = rawIntrospections.map(
-              ({ pgService, introspectionText }) => ({
+              ({
                 pgService,
+                introspectionText,
+                requiredSchemas,
+                catalogTypes,
+              }) => {
                 // IMPORTANT: parseIntrospectionResults must NOT be cached, because other plugins mutate it.
-                introspection: parseIntrospectionResults(introspectionText),
-              }),
+                const introspection =
+                  parseIntrospectionResults(introspectionText);
+                assertScopedNamespaces(
+                  introspection,
+                  requiredSchemas,
+                  pgService.name,
+                );
+                assertDependencyClosureTypes(
+                  introspection,
+                  catalogTypes,
+                  pgService.name,
+                );
+                return { pgService, introspection };
+              },
             );
 
             // Store the resolved state, so access during announcements doesn't cause the system to hang
@@ -779,7 +802,9 @@ export const PgIntrospectionPlugin: GraphileConfig.Plugin = {
 
 function introspectPgServices(
   pgServices: ReadonlyArray<GraphileConfig.PgServiceConfiguration> | undefined,
+  scopedIntrospection: GraphileBuild.GatherOptions["pgScopedIntrospection"],
 ): Promise<RawIntrospectionResults> {
+  assertScopedIntrospectionServices(pgServices, scopedIntrospection);
   if (!pgServices) {
     return Promise.resolve([]);
   }
@@ -835,21 +860,26 @@ function introspectPgServices(
       }
 
       // Do the introspection
-      const introspectionQuery = makeIntrospectionQuery();
+      const { query, requiredSchemas, catalogTypes } = getIntrospectionQuery(
+        pgService,
+        scopedIntrospection?.[name],
+      );
       const {
         rows: [row],
       } = await withPgClientFromPgService(
         pgService,
         pgService.pgSettingsForIntrospection ?? null,
-        (client) =>
-          client.query<{ introspection: string }>({
-            text: introspectionQuery,
-          }),
+        (client) => client.query<{ introspection: string }>(query),
       );
       if (!row) {
         throw new Error("Introspection failed");
       }
-      return { pgService, introspectionText: row.introspection };
+      return {
+        pgService,
+        introspectionText: row.introspection,
+        requiredSchemas,
+        catalogTypes,
+      };
     }),
   );
 }
diff --git a/graphile-build/graphile-build-pg/src/scopedIntrospection.ts b/graphile-build/graphile-build-pg/src/scopedIntrospection.ts
new file mode 100644
index 0000000000..10254f3268
--- /dev/null
+++ b/graphile-build/graphile-build-pg/src/scopedIntrospection.ts
@@ -0,0 +1,202 @@
+import type { Introspection, ScopedCatalogTypes } from "pg-introspection";
+import {
+  makeIntrospectionQuery,
+  makeSchemaScopedIntrospectionQuery,
+} from "pg-introspection";
+
+declare global {
+  namespace GraphileBuild {
+    interface GatherOptions {
+      /**
+       * Schema-scoped introspection options keyed by PostgreSQL service name.
+       * `true` enables defaults, `false` disables, and an object customizes it.
+       * Services without an entry continue to use stock introspection.
+       */
+      pgScopedIntrospection?: Readonly<
+        Record<string, PgScopedIntrospectionServiceConfig>
+      >;
+    }
+  }
+}
+
+export interface PgScopedIntrospectionOptions {
+  /** Controls how many `pg_catalog` types scoped introspection retains. */
+  catalogTypes?: ScopedCatalogTypes;
+
+  /**
+   * Extensions whose metadata should be retained even if no scoped object
+   * directly depends on them.
+   */
+  capabilityExtensions?: readonly string[];
+}
+
+export type PgScopedIntrospectionServiceConfig =
+  | boolean
+  | PgScopedIntrospectionOptions;
+
+export interface IntrospectionScope {
+  requiredSchemas: readonly string[] | null;
+  catalogTypes: ScopedCatalogTypes | null;
+}
+
+export interface IntrospectionQueryPlan extends IntrospectionScope {
+  query: { text: string; values?: unknown[] };
+}
+
+export function getIntrospectionQuery(
+  pgService: GraphileConfig.PgServiceConfiguration,
+  config?: PgScopedIntrospectionServiceConfig,
+): IntrospectionQueryPlan {
+  if (!config) {
+    return {
+      query: { text: makeIntrospectionQuery() },
+      requiredSchemas: null,
+      catalogTypes: null,
+    };
+  }
+
+  const options = config === true ? {} : config;
+
+  const requiredSchemas = pgService.schemas ?? [];
+  const catalogTypes = options.catalogTypes ?? "all";
+
+  return {
+    query: makeSchemaScopedIntrospectionQuery(requiredSchemas, {
+      catalogTypes,
+      capabilityExtensions: options.capabilityExtensions ?? [],
+    }),
+    requiredSchemas,
+    catalogTypes,
+  };
+}
+
+export function assertScopedIntrospectionServices(
+  pgServices: ReadonlyArray<GraphileConfig.PgServiceConfiguration> | undefined,
+  options: GraphileBuild.GatherOptions["pgScopedIntrospection"],
+): void {
+  if (!options) return;
+
+  const serviceNames = new Set(
+    (pgServices ?? []).map((pgService) => pgService.name),
+  );
+  const unknownServiceNames = Object.keys(options).filter(
+    (serviceName) => !serviceNames.has(serviceName),
+  );
+  if (unknownServiceNames.length > 0) {
+    throw new Error(
+      `Schema-scoped introspection configured for unknown PostgreSQL service(s): ${unknownServiceNames.join(
+        ", ",
+      )}`,
+    );
+  }
+}
+
+export function assertScopedNamespaces(
+  introspection: Introspection,
+  requiredSchemas: readonly string[] | null,
+  serviceName: string,
+): void {
+  if (requiredSchemas === null) return;
+
+  const found = new Set(
+    introspection.namespaces.map((namespace) => namespace.nspname),
+  );
+  const missing = requiredSchemas.filter((schema) => !found.has(schema));
+  if (missing.length > 0) {
+    throw new Error(
+      `Schema-scoped introspection for service '${serviceName}' did not find required schema(s): ${missing.join(
+        ", ",
+      )}`,
+    );
+  }
+}
+
+export function assertDependencyClosureTypes(
+  introspection: Introspection,
+  catalogTypes: ScopedCatalogTypes | null,
+  serviceName: string,
+): void {
+  if (catalogTypes !== "dependency-closure") return;
+
+  const retainedTypeOids = new Set(introspection.types.map((type) => type._id));
+  const requireType = (
+    oid: string | null | undefined,
+    objectKind: string,
+    objectContext: string,
+    field: string,
+  ): void => {
+    if (oid === null || oid === undefined || oid === "0") return;
+    // Extension-owned composite resources are removed from the public arrays
+    // after lookup hydration; the lookup remains available to consumers.
+    const introspectionLookups = (
+      introspection as Introspection & {
+        _lookups: { typeById: Map<string, unknown> };
+      }
+    )._lookups;
+    const resolves =
+      retainedTypeOids.has(oid) || introspectionLookups.typeById.has(oid);
+    if (!resolves) {
+      throw new Error(
+        `Dependency-closure introspection for service '${serviceName}' retained ${objectKind} '${objectContext}' field '${field}' referencing missing pg_type OID '${oid}'`,
+      );
+    }
+  };
+  const requireTypes = (
+    oids: readonly string[] | null | undefined,
+    objectKind: string,
+    objectContext: string,
+    field: string,
+  ): void => {
+    for (const oid of oids ?? []) {
+      requireType(oid, objectKind, objectContext, field);
+    }
+  };
+
+  for (const entity of introspection.classes) {
+    const context = `${entity.relname} (${entity._id})`;
+    requireType(entity.reltype, "pg_class", context, "reltype");
+    requireType(entity.reloftype, "pg_class", context, "reloftype");
+  }
+  for (const entity of introspection.attributes) {
+    requireType(
+      entity.atttypid,
+      "pg_attribute",
+      `${entity.attrelid}.${entity.attname}`,
+      "atttypid",
+    );
+  }
+  for (const entity of introspection.constraints) {
+    requireType(
+      entity.contypid,
+      "pg_constraint",
+      `${entity.conname} (${entity._id})`,
+      "contypid",
+    );
+  }
+  for (const entity of introspection.procs) {
+    const context = `${entity.proname} (${entity._id})`;
+    requireType(entity.prorettype, "pg_proc", context, "prorettype");
+    requireTypes(entity.proargtypes, "pg_proc", context, "proargtypes");
+    requireTypes(entity.proallargtypes, "pg_proc", context, "proallargtypes");
+  }
+  for (const entity of introspection.types) {
+    const context = `${entity.typname} (${entity._id})`;
+    requireType(entity.typbasetype, "pg_type", context, "typbasetype");
+    requireType(entity.typelem, "pg_type", context, "typelem");
+    requireType(entity.typarray, "pg_type", context, "typarray");
+  }
+  for (const entity of introspection.enums) {
+    requireType(
+      entity.enumtypid,
+      "pg_enum",
+      `${entity.enumlabel} (${entity._id})`,
+      "enumtypid",
+    );
+  }
+  for (const entity of introspection.ranges) {
+    const context = `range ${entity.rngtypid ?? "unknown"}`;
+    requireType(entity.rngtypid, "pg_range", context, "rngtypid");
+    requireType(entity.rngsubtype, "pg_range", context, "rngsubtype");
+    requireType(entity.rngmultitypid, "pg_range", context, "rngmultitypid");
+  }
+}
diff --git a/utils/pg-introspection/README.md b/utils/pg-introspection/README.md
index 8635b95aa5..90d93c7285 100644
--- a/utils/pg-introspection/README.md
+++ b/utils/pg-introspection/README.md
@@ -65,6 +65,39 @@ async function main() {
 main();
 ```
 
+### Schema-scoped introspection
+
+For databases with a large catalog, `makeSchemaScopedIntrospectionQuery()` can
+limit the result to objects in selected schemas and their transitive catalog
+dependencies:
+
+```js
+import {
+  makeSchemaScopedIntrospectionQuery,
+  parseIntrospectionResults,
+} from "pg-introspection";
+
+const query = makeSchemaScopedIntrospectionQuery(["app_public"], {
+  catalogTypes: "dependency-closure",
+  capabilityExtensions: ["pg_trgm"],
+});
+const { rows } = await pool.query(query);
+const introspection = parseIntrospectionResults(rows[0].introspection);
+```
+
+Schema and extension names are passed as query parameters. The dependency
+closure includes referenced relations, constraints, function signature types,
+domains, arrays, ranges, multiranges, indexes, inheritance parents, and
+extension metadata required by retained indexes. Dependencies cross schema
+boundaries automatically when a retained object needs them; unrelated objects
+are excluded.
+
+Extensions required by retained objects are also discovered automatically. Use
+`capabilityExtensions` when a consumer needs to know that an extension is
+installed even though no retained object depends on it. For example, a plugin
+can request `pg_trgm` metadata before registering an optional search capability.
+This retains the extension record, not every object owned by the extension.
+
 ## Accessors
 
 Into the introspection results we mix "accessor" functions to make following
diff --git a/utils/pg-introspection/__tests__/scoped-introspection-test.ts b/utils/pg-introspection/__tests__/scoped-introspection-test.ts
new file mode 100644
index 0000000000..219d7468bd
--- /dev/null
+++ b/utils/pg-introspection/__tests__/scoped-introspection-test.ts
@@ -0,0 +1,87 @@
+import assert from "node:assert/strict";
+import { createHash } from "node:crypto";
+import { describe, it } from "node:test";
+
+import {
+  makeIntrospectionQuery,
+  makeSchemaScopedIntrospectionQuery,
+} from "../src/index.ts";
+
+describe("schema-scoped introspection query", () => {
+  it("does not change the stock introspection query", () => {
+    // Exact query hash from before buildIntrospectionQuery was introduced.
+    const hash = createHash("sha256")
+      .update(makeIntrospectionQuery())
+      .digest("hex");
+    assert.equal(
+      hash,
+      "c0ed817b912f78e1ea68c70d89ff4b7f9cb4c02d88112a69ac4109d5b996e4c5",
+    );
+  });
+
+  it("keeps schema and extension names in query parameters", () => {
+    const schema = "tenant_a'); drop schema public; --";
+    const extension = "pg_trgm'); select pg_sleep(10); --";
+    const query = makeSchemaScopedIntrospectionQuery(
+      [schema, "tenant_a", schema],
+      { capabilityExtensions: [extension, "pg_trgm", extension] },
+    );
+
+    assert.match(query.text, /pg_catalog\.unnest\(\$1::text\[\]\)/);
+    assert.match(query.text, /pg_catalog\.unnest\(\$2::text\[\]\)/);
+    assert.equal(query.text.includes(schema), false);
+    assert.equal(query.text.includes(extension), false);
+    assert.deepEqual(query.values, [
+      [schema, "tenant_a"],
+      [extension, "pg_trgm"],
+    ]);
+  });
+
+  it("rejects invalid schema and extension names", () => {
+    assert.throws(
+      () => makeSchemaScopedIntrospectionQuery([]),
+      /requires at least one schema/,
+    );
+    assert.throws(
+      () => makeSchemaScopedIntrospectionQuery(["pg_catalog"]),
+      /cannot expose system schema 'pg_catalog'/,
+    );
+    assert.throws(
+      () => makeSchemaScopedIntrospectionQuery(["information_schema"]),
+      /cannot expose system schema 'information_schema'/,
+    );
+    assert.throws(
+      () => makeSchemaScopedIntrospectionQuery(["tenant\0a"]),
+      /must not contain NUL bytes/,
+    );
+    assert.throws(
+      () =>
+        makeSchemaScopedIntrospectionQuery(["tenant_a"], {
+          capabilityExtensions: [" pg_trgm"],
+        }),
+      /must contain exact non-empty extension names/,
+    );
+  });
+
+  it("supports full and dependency-closure catalog type policies", () => {
+    const all = makeSchemaScopedIntrospectionQuery(["tenant_a"]);
+    const closure = makeSchemaScopedIntrospectionQuery(["tenant_a"], {
+      catalogTypes: "dependency-closure",
+    });
+
+    for (const query of [all, closure]) {
+      assert.match(query.text, /with\nrecursive/u);
+      assert.match(query.text, /object_closure\(object_class, object_id\) as/u);
+      assert.match(query.text, /retained_index_support_objects/u);
+      assert.match(query.text, /installed_extensions/u);
+    }
+    assert.match(
+      all.text,
+      /or pg_type\.typnamespace = 'pg_catalog'::regnamespace/u,
+    );
+    assert.doesNotMatch(
+      closure.text,
+      /or pg_type\.typnamespace = 'pg_catalog'::regnamespace/u,
+    );
+  });
+});
diff --git a/utils/pg-introspection/src/index.ts b/utils/pg-introspection/src/index.ts
index 903a49d569..e2e800c354 100644
--- a/utils/pg-introspection/src/index.ts
+++ b/utils/pg-introspection/src/index.ts
@@ -22,6 +22,12 @@ import type {
   PgType,
 } from "./introspection.ts";
 export { makeIntrospectionQuery } from "./introspection.ts";
+export {
+  makeSchemaScopedIntrospectionQuery,
+  type SchemaScopedIntrospectionOptions,
+  type SchemaScopedIntrospectionQuery,
+  type ScopedCatalogTypes,
+} from "./scopedIntrospection.ts";
 import type { AclObject } from "./acl.ts";
 import {
   aclContainsRole,
diff --git a/utils/pg-introspection/src/introspection.ts b/utils/pg-introspection/src/introspection.ts
index b1e6075a24..7f0e551134 100644
--- a/utils/pg-introspection/src/introspection.ts
+++ b/utils/pg-introspection/src/introspection.ts
@@ -1570,12 +1570,35 @@ export type PgEntity =
   | PgDescription
   | PgAm;
 
+export interface IntrospectionQueryScope {
+  ctes?: string;
+  namespacePredicate: string;
+  classPredicate: string;
+  constraintPredicate: string;
+  procPredicate: string;
+  typePredicate: string;
+  extensionPredicate?: string;
+}
+
+const STOCK_QUERY_SCOPE: IntrospectionQueryScope = {
+  namespacePredicate: "nspname <> 'information_schema'",
+  classPredicate:
+    "relnamespace in (select namespaces._id from namespaces where nspname <> 'information_schema' and nspname not like 'pg\\_%')",
+  constraintPredicate:
+    "connamespace in (select namespaces._id from namespaces where nspname <> 'information_schema' and nspname not like 'pg\\_%')",
+  procPredicate:
+    "pronamespace in (select namespaces._id from namespaces where nspname <> 'information_schema' and nspname not like 'pg\\_%')",
+  typePredicate:
+    "(typnamespace in (select namespaces._id from namespaces where nspname <> 'information_schema' and nspname not like 'pg\\_%'))\n    or (typnamespace = 'pg_catalog'::regnamespace)",
+};
+
 // We might want this to take options in future, so we've made it a function.
 /**
  * Builds a PostgreSQL introspection SQL query to return an object with the same shape as `Introspection` above.
  */
-export const makeIntrospectionQuery = () => `\
+export const buildIntrospectionQuery = (scope: IntrospectionQueryScope) => `\
 with
+${scope.ctes ?? ""}\
   database as (
     select pg_database.oid as _id, *
     from pg_catalog.pg_database
@@ -1585,14 +1608,14 @@ with
   namespaces as (
     select pg_namespace.oid as _id, *
     from pg_catalog.pg_namespace
-    where nspname <> 'information_schema'
+    where ${scope.namespacePredicate}
   ),
 
   classes as (
     select pg_class.oid as _id, *,
       pg_catalog.pg_relation_is_updatable(oid, true)::bit(8)::int4 as "updatable_mask"
     from pg_catalog.pg_class
-    where relnamespace in (select namespaces._id from namespaces where nspname <> 'information_schema' and nspname not like 'pg\\_%')
+    where ${scope.classPredicate}
   ),
 
   attributes as (
@@ -1604,13 +1627,13 @@ with
   constraints as (
     select pg_constraint.oid as _id, *
     from pg_catalog.pg_constraint
-    where connamespace in (select namespaces._id from namespaces where nspname <> 'information_schema' and nspname not like 'pg\\_%')
+    where ${scope.constraintPredicate}
   ),
 
   procs as (
     select pg_proc.oid as _id, *
     from pg_catalog.pg_proc
-    where pronamespace in (select namespaces._id from namespaces where nspname <> 'information_schema' and nspname not like 'pg\\_%')
+    where ${scope.procPredicate}
     and prorettype operator(pg_catalog.<>) 2279
   ),
 
@@ -1628,8 +1651,7 @@ with
   types as (
     select pg_type.oid as _id, *
     from pg_catalog.pg_type
-    where (typnamespace in (select namespaces._id from namespaces where nspname <> 'information_schema' and nspname not like 'pg\\_%'))
-    or (typnamespace = 'pg_catalog'::regnamespace)
+    where ${scope.typePredicate}
   ),
 
   enums as (
@@ -1641,6 +1663,12 @@ with
   extensions as (
     select pg_extension.oid as _id, *
     from pg_catalog.pg_extension
+${
+  scope.extensionPredicate
+    ? `    where ${scope.extensionPredicate}
+`
+    : ""
+}\
   ),
 
   indexes as (
@@ -1785,3 +1813,6 @@ select json_build_object(
   1
 )::text as introspection
 `;
+
+export const makeIntrospectionQuery = () =>
+  buildIntrospectionQuery(STOCK_QUERY_SCOPE);
diff --git a/utils/pg-introspection/src/scopedIntrospection.ts b/utils/pg-introspection/src/scopedIntrospection.ts
new file mode 100644
index 0000000000..ab74c519c2
--- /dev/null
+++ b/utils/pg-introspection/src/scopedIntrospection.ts
@@ -0,0 +1,413 @@
+import { buildIntrospectionQuery } from "./introspection.ts";
+
+export type ScopedCatalogTypes = "all" | "dependency-closure";
+
+export interface SchemaScopedIntrospectionOptions {
+  catalogTypes?: ScopedCatalogTypes;
+  capabilityExtensions?: readonly string[];
+}
+
+export interface SchemaScopedIntrospectionQuery {
+  text: string;
+  values: [string[], string[]];
+}
+
+const SCOPED_CTES = `recursive
+  requested_schema_names(schema_name) as (
+    select distinct requested.schema_name
+    from pg_catalog.unnest($1::text[]) as requested(schema_name)
+  ),
+
+  capability_extension_names(extension_name) as (
+    select distinct capability.extension_name
+    from pg_catalog.unnest($2::text[]) as capability(extension_name)
+  ),
+
+  requested_namespaces as (
+    select pg_namespace.oid as _id, pg_namespace.nspname
+    from pg_catalog.pg_namespace
+    inner join requested_schema_names
+      on requested_schema_names.schema_name = pg_namespace.nspname
+  ),
+
+  root_objects(object_class, object_id) as (
+    select 'pg_catalog.pg_class'::regclass::oid, pg_class.oid
+    from pg_catalog.pg_class
+    where pg_class.relnamespace in (select requested_namespaces._id from requested_namespaces)
+
+    union
+
+    select 'pg_catalog.pg_constraint'::regclass::oid, pg_constraint.oid
+    from pg_catalog.pg_constraint
+    where pg_constraint.connamespace in (select requested_namespaces._id from requested_namespaces)
+
+    union
+
+    select 'pg_catalog.pg_proc'::regclass::oid, pg_proc.oid
+    from pg_catalog.pg_proc
+    where pg_proc.pronamespace in (select requested_namespaces._id from requested_namespaces)
+      and pg_proc.prorettype operator(pg_catalog.<>) 2279
+
+    union
+
+    select 'pg_catalog.pg_type'::regclass::oid, pg_type.oid
+    from pg_catalog.pg_type
+    where pg_type.typnamespace in (select requested_namespaces._id from requested_namespaces)
+  ),
+
+  object_closure(object_class, object_id) as (
+    select root_objects.object_class, root_objects.object_id
+    from root_objects
+
+    union
+
+    select dependency.object_class, dependency.object_id
+    from object_closure
+    cross join lateral (
+      select
+        'pg_catalog.pg_type'::regclass::oid as object_class,
+        pg_class.reltype as object_id
+      from pg_catalog.pg_class
+      where object_closure.object_class = 'pg_catalog.pg_class'::regclass
+        and pg_class.oid = object_closure.object_id
+
+      union all
+
+      select 'pg_catalog.pg_type'::regclass::oid, pg_class.reloftype
+      from pg_catalog.pg_class
+      where object_closure.object_class = 'pg_catalog.pg_class'::regclass
+        and pg_class.oid = object_closure.object_id
+
+      union all
+
+      select 'pg_catalog.pg_type'::regclass::oid, pg_attribute.atttypid
+      from pg_catalog.pg_attribute
+      where object_closure.object_class = 'pg_catalog.pg_class'::regclass
+        and pg_attribute.attrelid = object_closure.object_id
+
+      union all
+
+      select 'pg_catalog.pg_constraint'::regclass::oid, pg_constraint.oid
+      from pg_catalog.pg_constraint
+      where object_closure.object_class = 'pg_catalog.pg_class'::regclass
+        and pg_constraint.conrelid = object_closure.object_id
+
+      union all
+
+      select 'pg_catalog.pg_class'::regclass::oid, pg_index.indexrelid
+      from pg_catalog.pg_index
+      where object_closure.object_class = 'pg_catalog.pg_class'::regclass
+        and pg_index.indrelid = object_closure.object_id
+
+      union all
+
+      select 'pg_catalog.pg_class'::regclass::oid, pg_inherits.inhparent
+      from pg_catalog.pg_inherits
+      where object_closure.object_class = 'pg_catalog.pg_class'::regclass
+        and pg_inherits.inhrelid = object_closure.object_id
+
+      union all
+
+      select 'pg_catalog.pg_class'::regclass::oid, constraint_class.oid
+      from pg_catalog.pg_constraint
+      cross join lateral pg_catalog.unnest(
+        array[
+          pg_constraint.conrelid,
+          pg_constraint.confrelid,
+          pg_constraint.conindid
+        ]::oid[]
+      ) as constraint_class(oid)
+      where object_closure.object_class = 'pg_catalog.pg_constraint'::regclass
+        and pg_constraint.oid = object_closure.object_id
+
+      union all
+
+      select 'pg_catalog.pg_type'::regclass::oid, pg_constraint.contypid
+      from pg_catalog.pg_constraint
+      where object_closure.object_class = 'pg_catalog.pg_constraint'::regclass
+        and pg_constraint.oid = object_closure.object_id
+
+      union all
+
+      select 'pg_catalog.pg_constraint'::regclass::oid, pg_constraint.conparentid
+      from pg_catalog.pg_constraint
+      where object_closure.object_class = 'pg_catalog.pg_constraint'::regclass
+        and pg_constraint.oid = object_closure.object_id
+
+      union all
+
+      select 'pg_catalog.pg_type'::regclass::oid, procedure_type.oid
+      from pg_catalog.pg_proc
+      cross join lateral pg_catalog.unnest(
+        coalesce(pg_proc.proallargtypes, pg_proc.proargtypes::oid[])
+        || array[pg_proc.prorettype]::oid[]
+      ) as procedure_type(oid)
+      where object_closure.object_class = 'pg_catalog.pg_proc'::regclass
+        and pg_proc.oid = object_closure.object_id
+
+      union all
+
+      select 'pg_catalog.pg_type'::regclass::oid, dependency_type.oid
+      from pg_catalog.pg_type
+      cross join lateral pg_catalog.unnest(
+        array[
+          pg_type.typbasetype,
+          pg_type.typelem,
+          pg_type.typarray
+        ]::oid[]
+      ) as dependency_type(oid)
+      where object_closure.object_class = 'pg_catalog.pg_type'::regclass
+        and pg_type.oid = object_closure.object_id
+
+      union all
+
+      select 'pg_catalog.pg_class'::regclass::oid, pg_type.typrelid
+      from pg_catalog.pg_type
+      where object_closure.object_class = 'pg_catalog.pg_type'::regclass
+        and pg_type.oid = object_closure.object_id
+
+      union all
+
+      select 'pg_catalog.pg_constraint'::regclass::oid, pg_constraint.oid
+      from pg_catalog.pg_constraint
+      where object_closure.object_class = 'pg_catalog.pg_type'::regclass
+        and pg_constraint.contypid = object_closure.object_id
+
+      union all
+
+      select 'pg_catalog.pg_type'::regclass::oid, range_type.oid
+      from pg_catalog.pg_range
+      cross join lateral pg_catalog.unnest(
+        array[
+          pg_range.rngtypid,
+          pg_range.rngsubtype,
+          pg_range.rngmultitypid
+        ]::oid[]
+      ) as range_type(oid)
+      where object_closure.object_class = 'pg_catalog.pg_type'::regclass
+        and object_closure.object_id in (pg_range.rngtypid, pg_range.rngmultitypid)
+    ) as dependency
+    where dependency.object_id operator(pg_catalog.<>) 0
+  ),
+
+  retained_index_metadata(indexrelid, indclass, indcollation) as (
+    select pg_index.indexrelid, pg_index.indclass, pg_index.indcollation
+    from object_closure
+    inner join pg_catalog.pg_class retained_index
+      on object_closure.object_class = 'pg_catalog.pg_class'::regclass
+      and retained_index.oid = object_closure.object_id
+      and retained_index.relkind in ('i', 'I')
+    inner join pg_catalog.pg_index
+      on pg_index.indexrelid = retained_index.oid
+  ),
+
+  retained_index_opclasses(_id, opcfamily) as (
+    select pg_opclass.oid, pg_opclass.opcfamily
+    from retained_index_metadata
+    cross join lateral pg_catalog.unnest(
+      retained_index_metadata.indclass::oid[]
+    ) as index_opclass(_id)
+    inner join pg_catalog.pg_opclass
+      on pg_opclass.oid = index_opclass._id
+  ),
+
+  retained_index_support_objects(object_class, object_id) as (
+    select 'pg_catalog.pg_opclass'::regclass::oid, retained_index_opclasses._id
+    from retained_index_opclasses
+
+    union
+
+    select 'pg_catalog.pg_opfamily'::regclass::oid, retained_index_opclasses.opcfamily
+    from retained_index_opclasses
+
+    union
+
+    select 'pg_catalog.pg_operator'::regclass::oid, pg_amop.amopopr
+    from retained_index_opclasses
+    inner join pg_catalog.pg_amop
+      on pg_amop.amopfamily = retained_index_opclasses.opcfamily
+
+    union
+
+    select 'pg_catalog.pg_proc'::regclass::oid, pg_amproc.amproc
+    from retained_index_opclasses
+    inner join pg_catalog.pg_amproc
+      on pg_amproc.amprocfamily = retained_index_opclasses.opcfamily
+
+    union
+
+    select 'pg_catalog.pg_collation'::regclass::oid, index_collation._id
+    from retained_index_metadata
+    cross join lateral pg_catalog.unnest(
+      retained_index_metadata.indcollation::oid[]
+    ) as index_collation(_id)
+    where index_collation._id operator(pg_catalog.<>) 0
+  ),
+
+  installed_extensions(_id, extnamespace) as (
+    select pg_extension.oid, pg_extension.extnamespace
+    from pg_catalog.pg_extension
+    where pg_extension.extname in (
+      select capability_extension_names.extension_name
+      from capability_extension_names
+    )
+    or exists (
+      select 1
+      from object_closure
+      inner join pg_catalog.pg_depend
+        on pg_depend.classid = object_closure.object_class
+        and pg_depend.objid = object_closure.object_id
+        and pg_depend.refclassid = 'pg_catalog.pg_extension'::regclass
+        and pg_depend.refobjid = pg_extension.oid
+        and pg_depend.deptype = 'e'
+    )
+    or exists (
+      select 1
+      from retained_index_support_objects
+      inner join pg_catalog.pg_depend
+        on pg_depend.classid = retained_index_support_objects.object_class
+        and pg_depend.objid = retained_index_support_objects.object_id
+        and pg_depend.refclassid = 'pg_catalog.pg_extension'::regclass
+        and pg_depend.refobjid = pg_extension.oid
+        and pg_depend.deptype = 'e'
+    )
+    or exists (
+      select 1
+      from object_closure
+      inner join pg_catalog.pg_class retained_index
+        on object_closure.object_class = 'pg_catalog.pg_class'::regclass
+        and retained_index.oid = object_closure.object_id
+        and retained_index.relkind = 'i'
+      inner join pg_catalog.pg_depend
+        on pg_depend.classid = 'pg_catalog.pg_am'::regclass
+        and pg_depend.objid = retained_index.relam
+        and pg_depend.refclassid = 'pg_catalog.pg_extension'::regclass
+        and pg_depend.refobjid = pg_extension.oid
+        and pg_depend.deptype = 'e'
+    )
+  ),
+
+  scoped_namespaces(_id) as (
+    select requested_namespaces._id
+    from requested_namespaces
+
+    union
+
+    select pg_class.relnamespace
+    from object_closure
+    inner join pg_catalog.pg_class
+      on object_closure.object_class = 'pg_catalog.pg_class'::regclass
+      and pg_class.oid = object_closure.object_id
+
+    union
+
+    select pg_constraint.connamespace
+    from object_closure
+    inner join pg_catalog.pg_constraint
+      on object_closure.object_class = 'pg_catalog.pg_constraint'::regclass
+      and pg_constraint.oid = object_closure.object_id
+
+    union
+
+    select pg_proc.pronamespace
+    from object_closure
+    inner join pg_catalog.pg_proc
+      on object_closure.object_class = 'pg_catalog.pg_proc'::regclass
+      and pg_proc.oid = object_closure.object_id
+
+    union
+
+    select pg_type.typnamespace
+    from object_closure
+    inner join pg_catalog.pg_type
+      on object_closure.object_class = 'pg_catalog.pg_type'::regclass
+      and pg_type.oid = object_closure.object_id
+
+    union
+
+    select installed_extensions.extnamespace
+    from installed_extensions
+    where installed_extensions.extnamespace operator(pg_catalog.<>) 0
+
+    union
+
+    select pg_namespace.oid
+    from pg_catalog.pg_namespace
+    where pg_namespace.nspname = 'pg_catalog'
+  ),
+
+`;
+/**
+ * Builds a parameterized introspection query scoped to the requested schemas
+ * and the transitive object dependencies required by their objects.
+ */
+export const makeSchemaScopedIntrospectionQuery = (
+  schemas: readonly string[],
+  options: SchemaScopedIntrospectionOptions = {},
+): SchemaScopedIntrospectionQuery => {
+  if (schemas.length === 0) {
+    throw new Error("Schema-scoped introspection requires at least one schema");
+  }
+  const catalogTypes = options.catalogTypes ?? "all";
+  const capabilityExtensions = options.capabilityExtensions ?? [];
+  const normalizedCapabilityExtensions = Array.from(
+    new Set(
+      capabilityExtensions.map((extension) => {
+        if (
+          extension.length === 0 ||
+          extension.trim() !== extension ||
+          extension.includes("\0")
+        ) {
+          throw new Error(
+            "Schema-scoped introspection capabilityExtensions must contain exact non-empty extension names",
+          );
+        }
+        return extension;
+      }),
+    ),
+  );
+  const normalized = Array.from(
+    new Set(
+      schemas.map((schema) => {
+        if (schema.length === 0) {
+          throw new Error(
+            "Schema-scoped introspection schemas must be non-empty strings",
+          );
+        }
+        if (schema.includes("\0")) {
+          throw new Error(
+            "Schema-scoped introspection schemas must not contain NUL bytes",
+          );
+        }
+        if (schema === "information_schema" || schema.startsWith("pg_")) {
+          throw new Error(
+            `Schema-scoped introspection cannot expose system schema '${schema}'`,
+          );
+        }
+        return schema;
+      }),
+    ),
+  );
+  const dependencyClosureTypePredicate =
+    "pg_type.oid = any (array(select object_id from object_closure where object_class = 'pg_catalog.pg_type'::regclass))";
+  return {
+    text: buildIntrospectionQuery({
+      ctes: SCOPED_CTES,
+      namespacePredicate:
+        "pg_namespace.oid = any (array(select scoped_namespaces._id from scoped_namespaces))",
+      classPredicate:
+        "pg_class.oid = any (array(select object_id from object_closure where object_class = 'pg_catalog.pg_class'::regclass))",
+      constraintPredicate:
+        "pg_constraint.oid = any (array(select object_id from object_closure where object_class = 'pg_catalog.pg_constraint'::regclass))",
+      procPredicate:
+        "pg_proc.oid = any (array(select object_id from object_closure where object_class = 'pg_catalog.pg_proc'::regclass))",
+      typePredicate:
+        catalogTypes === "all"
+          ? `${dependencyClosureTypePredicate} or pg_type.typnamespace = 'pg_catalog'::regnamespace`
+          : dependencyClosureTypePredicate,
+      extensionPredicate:
+        "pg_extension.oid = any (array(select installed_extensions._id from installed_extensions))",
+    }),
+    values: [normalized, normalizedCapabilityExtensions],
+  };
+};