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..119dbdec64 100644
--- a/graphile-build/graphile-build-pg/README.md
+++ b/graphile-build/graphile-build-pg/README.md
@@ -16,6 +16,62 @@ 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.
+
+`pg-introspection` owns the scoped query plan and validates the parsed result
+against that same plan. `graphile-build-pg` only selects stock or scoped mode
+for each service, executes the query, and adds service context to errors.
+
+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..afc5bc06dd
--- /dev/null
+++ b/graphile-build/graphile-build-pg/__tests__/scopedIntrospection.integration.test.ts
@@ -0,0 +1,292 @@
+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);
+  });
+
+  it("fails fast when a configured root schema is missing", async () => {
+    await expect(buildSchema(pool, true, "scope_missing_root")).rejects.toThrow(
+      /validation failed.*did not find required schema\(s\): scope_missing_root/u,
+    );
+  });
+});
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..4c140bf227
--- /dev/null
+++ b/graphile-build/graphile-build-pg/__tests__/scopedIntrospection.test.ts
@@ -0,0 +1,105 @@
+import { gather } from "graphile-build";
+import type { SchemaScopedIntrospectionOptions } from "pg-introspection";
+import { makeIntrospectionQuery } from "pg-introspection";
+
+import { PgIntrospectionPlugin } from "../src/index.ts";
+
+interface IntrospectionQuery {
+  text: string;
+  values?: unknown[];
+}
+
+type ServiceConfig = boolean | SchemaScopedIntrospectionOptions;
+
+async function captureIntrospectionQuery(
+  configByService?: Readonly<Record<string, ServiceConfig>>,
+): Promise<IntrospectionQuery> {
+  let capturedQuery: IntrospectionQuery | undefined;
+  const queryCaptured = new Error("query captured");
+  const pgService = {
+    name: "main",
+    schemas: ["app_public"],
+    withPgClientKey: "withPgClient",
+    pgSettingsKey: "pgSettings",
+    adaptorSettings: {},
+    adaptor: {
+      createWithPgClient() {
+        return async (
+          _pgSettings: Record<string, string | undefined> | null,
+          callback: (client: never) => Promise<unknown>,
+        ) =>
+          callback({
+            query(query: IntrospectionQuery) {
+              capturedQuery = query;
+              throw queryCaptured;
+            },
+          } as never);
+      },
+    },
+  } as GraphileConfig.PgServiceConfiguration;
+  const IntrospectionConsumerPlugin: GraphileConfig.Plugin = {
+    name: "IntrospectionConsumerPlugin",
+    after: ["PgIntrospectionPlugin"],
+    gather: {
+      async main(_output, info) {
+        await info.helpers.pgIntrospection.getIntrospection();
+      },
+    },
+  };
+
+  try {
+    await gather({
+      plugins: [PgIntrospectionPlugin, IntrospectionConsumerPlugin],
+      pgServices: [pgService],
+      ...(configByService
+        ? { gather: { pgScopedIntrospection: configByService } }
+        : null),
+    });
+  } catch (error) {
+    if (error !== queryCaptured) throw error;
+  }
+
+  if (!capturedQuery) {
+    throw new Error("PostgreSQL introspection query was not executed");
+  }
+  return capturedQuery;
+}
+
+describe("scoped introspection service configuration", () => {
+  it.each([undefined, false])(
+    "uses stock introspection for %p",
+    async (config) => {
+      await expect(
+        captureIntrospectionQuery(
+          config === undefined ? undefined : { main: config },
+        ),
+      ).resolves.toEqual({ text: makeIntrospectionQuery() });
+    },
+  );
+
+  it("uses scoped introspection defaults for true", async () => {
+    const query = await captureIntrospectionQuery({ main: true });
+
+    expect(query.values).toEqual([["app_public"], []]);
+  });
+
+  it("builds a scoped, parameterized query from the service schemas", async () => {
+    const query = await captureIntrospectionQuery({
+      main: {
+        catalogTypes: "dependency-closure",
+        capabilityExtensions: ["pg_trgm"],
+      },
+    });
+
+    expect(query.text).not.toContain(
+      "or pg_type.typnamespace = 'pg_catalog'::regnamespace",
+    );
+    expect(query.values).toEqual([["app_public"], ["pg_trgm"]]);
+  });
+
+  it("rejects configuration for an unknown PostgreSQL service", async () => {
+    await expect(captureIntrospectionQuery({ analytics: {} })).rejects.toThrow(
+      /unknown PostgreSQL service\(s\): analytics/u,
+    );
+  });
+});
diff --git a/graphile-build/graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts b/graphile-build/graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts
index bfab4a8cc9..13bd4371bd 100644
--- a/graphile-build/graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts
+++ b/graphile-build/graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts
@@ -26,10 +26,14 @@ import type {
   PgRange,
   PgRoles,
   PgType,
+  SchemaScopedIntrospectionOptions,
+  SchemaScopedIntrospectionPlan,
 } from "pg-introspection";
 import {
   makeIntrospectionQuery,
+  makeSchemaScopedIntrospectionPlan,
   parseIntrospectionResults,
+  validateSchemaScopedIntrospection,
 } from "pg-introspection";
 
 import { version } from "../version.ts";
@@ -54,6 +58,17 @@ export type PgEntityWithId =
   | PgIndex
   | PgLanguage;
 
+type PgScopedIntrospectionOptions = SchemaScopedIntrospectionOptions;
+
+type PgScopedIntrospectionServiceConfig =
+  | boolean
+  | PgScopedIntrospectionOptions;
+
+interface IntrospectionQueryPlan {
+  query: { text: string; values?: unknown[] };
+  scopedPlan: SchemaScopedIntrospectionPlan | null;
+}
+
 declare global {
   namespace GraphileBuild {
     interface GatherOptions {
@@ -63,6 +78,15 @@ declare global {
        * Default: true
        */
       installWatchFixtures?: boolean;
+
+      /**
+       * 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>
+      >;
     }
   }
 
@@ -242,9 +266,54 @@ declare global {
   }
 }
 
+function getIntrospectionQuery(
+  pgService: GraphileConfig.PgServiceConfiguration,
+  config?: PgScopedIntrospectionServiceConfig,
+): IntrospectionQueryPlan {
+  if (!config) {
+    return {
+      query: { text: makeIntrospectionQuery() },
+      scopedPlan: null,
+    };
+  }
+
+  const options = config === true ? {} : config;
+  const scopedPlan = makeSchemaScopedIntrospectionPlan(
+    pgService.schemas ?? [],
+    options,
+  );
+
+  return {
+    query: scopedPlan.query,
+    scopedPlan,
+  };
+}
+
+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(
+        ", ",
+      )}`,
+    );
+  }
+}
+
 type RawIntrospectionResults = Array<{
   pgService: GraphileConfig.PgServiceConfiguration;
   introspectionText: string;
+  scopedPlan: SchemaScopedIntrospectionPlan | null;
 }>;
 type IntrospectionResults = Array<{
   pgService: GraphileConfig.PgServiceConfiguration;
@@ -534,6 +603,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 +614,26 @@ export const PgIntrospectionPlugin: GraphileConfig.Plugin = {
             const rawIntrospections = await introspectionPromise;
 
             const introspections: IntrospectionResults = rawIntrospections.map(
-              ({ pgService, introspectionText }) => ({
-                pgService,
+              ({ pgService, introspectionText, scopedPlan }) => {
                 // IMPORTANT: parseIntrospectionResults must NOT be cached, because other plugins mutate it.
-                introspection: parseIntrospectionResults(introspectionText),
-              }),
+                const introspection =
+                  parseIntrospectionResults(introspectionText);
+                if (scopedPlan) {
+                  try {
+                    validateSchemaScopedIntrospection(
+                      introspection,
+                      scopedPlan,
+                    );
+                  } catch (error) {
+                    const message =
+                      error instanceof Error ? error.message : String(error);
+                    throw new Error(
+                      `Schema-scoped introspection validation failed for PostgreSQL service '${pgService.name}': ${message}`,
+                    );
+                  }
+                }
+                return { pgService, introspection };
+              },
             );
 
             // Store the resolved state, so access during announcements doesn't cause the system to hang
@@ -779,7 +864,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 +922,25 @@ function introspectPgServices(
       }
 
       // Do the introspection
-      const introspectionQuery = makeIntrospectionQuery();
+      const { query, scopedPlan } = 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,
+        scopedPlan,
+      };
     }),
   );
 }
diff --git a/utils/pg-introspection/README.md b/utils/pg-introspection/README.md
index 8635b95aa5..ece4b38342 100644
--- a/utils/pg-introspection/README.md
+++ b/utils/pg-introspection/README.md
@@ -65,6 +65,41 @@ 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 {
+  makeSchemaScopedIntrospectionPlan,
+  parseIntrospectionResults,
+  validateSchemaScopedIntrospection,
+} from "pg-introspection";
+
+const plan = makeSchemaScopedIntrospectionPlan(["app_public"], {
+  catalogTypes: "dependency-closure",
+  capabilityExtensions: ["pg_trgm"],
+});
+const { rows } = await pool.query(plan.query);
+const introspection = parseIntrospectionResults(rows[0].introspection);
+validateSchemaScopedIntrospection(introspection, plan);
+```
+
+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..196881e082
--- /dev/null
+++ b/utils/pg-introspection/__tests__/scoped-introspection-test.ts
@@ -0,0 +1,204 @@
+import assert from "node:assert/strict";
+import { createHash } from "node:crypto";
+import { describe, it } from "node:test";
+
+import {
+  type Introspection,
+  makeIntrospectionQuery,
+  makeSchemaScopedIntrospectionPlan,
+  makeSchemaScopedIntrospectionQuery,
+  validateSchemaScopedIntrospection,
+} from "../src/index.ts";
+
+function makeIntrospection(
+  overrides: Partial<Introspection> = {},
+  lookupTypeOids: readonly string[] = [],
+): Introspection {
+  return {
+    namespaces: [],
+    classes: [],
+    attributes: [],
+    constraints: [],
+    procs: [],
+    types: [],
+    enums: [],
+    ranges: [],
+    _lookups: {
+      typeById: new Map(lookupTypeOids.map((oid) => [oid, {}])),
+    },
+    ...overrides,
+  } as unknown as Introspection;
+}
+
+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,
+    );
+  });
+
+  it("shares normalized scope data between query and validation", () => {
+    const plan = makeSchemaScopedIntrospectionPlan(
+      ["app_public", "app_public"],
+      {
+        catalogTypes: "dependency-closure",
+        capabilityExtensions: ["pg_trgm", "pg_trgm"],
+      },
+    );
+
+    assert.deepEqual(plan.scope, {
+      schemas: ["app_public"],
+      catalogTypes: "dependency-closure",
+      capabilityExtensions: ["pg_trgm"],
+    });
+    assert.equal(plan.query.values[0], plan.scope.schemas);
+    assert.equal(plan.query.values[1], plan.scope.capabilityExtensions);
+  });
+
+  it("fails fast when a required root schema is missing", () => {
+    const plan = makeSchemaScopedIntrospectionPlan(["app_public"]);
+
+    assert.throws(
+      () => validateSchemaScopedIntrospection(makeIntrospection(), plan),
+      /did not find required schema\(s\): app_public/u,
+    );
+  });
+
+  it("fails fast when a function dependency type is missing", () => {
+    const plan = makeSchemaScopedIntrospectionPlan(["app_public"], {
+      catalogTypes: "dependency-closure",
+    });
+    const introspection = makeIntrospection({
+      namespaces: [{ nspname: "app_public" }] as Introspection["namespaces"],
+      procs: [
+        {
+          _id: "20",
+          proname: "missing_result",
+          prorettype: "999",
+          proargtypes: [],
+          proallargtypes: null,
+        },
+      ] as Introspection["procs"],
+    });
+
+    assert.throws(
+      () => validateSchemaScopedIntrospection(introspection, plan),
+      /pg_proc.*prorettype.*missing pg_type OID '999'/u,
+    );
+  });
+
+  it("fails fast on a dangling column type", () => {
+    const plan = makeSchemaScopedIntrospectionPlan(["app_public"], {
+      catalogTypes: "dependency-closure",
+    });
+    const introspection = makeIntrospection({
+      namespaces: [{ nspname: "app_public" }] as Introspection["namespaces"],
+      attributes: [
+        {
+          attrelid: "10",
+          attname: "dangling_value",
+          atttypid: "999",
+        },
+      ] as Introspection["attributes"],
+    });
+
+    assert.throws(
+      () => validateSchemaScopedIntrospection(introspection, plan),
+      /pg_attribute.*atttypid.*missing pg_type OID '999'/u,
+    );
+  });
+
+  it("accepts dependency types retained in internal lookups", () => {
+    const plan = makeSchemaScopedIntrospectionPlan(["app_public"], {
+      catalogTypes: "dependency-closure",
+    });
+    const introspection = makeIntrospection(
+      {
+        namespaces: [{ nspname: "app_public" }] as Introspection["namespaces"],
+        attributes: [
+          {
+            attrelid: "10",
+            attname: "extension_value",
+            atttypid: "999",
+          },
+        ] as Introspection["attributes"],
+      },
+      ["999"],
+    );
+
+    assert.doesNotThrow(() =>
+      validateSchemaScopedIntrospection(introspection, plan),
+    );
+  });
+});
diff --git a/utils/pg-introspection/src/index.ts b/utils/pg-introspection/src/index.ts
index 903a49d569..3e4b27a514 100644
--- a/utils/pg-introspection/src/index.ts
+++ b/utils/pg-introspection/src/index.ts
@@ -22,6 +22,16 @@ import type {
   PgType,
 } from "./introspection.ts";
 export { makeIntrospectionQuery } from "./introspection.ts";
+export {
+  makeSchemaScopedIntrospectionPlan,
+  makeSchemaScopedIntrospectionQuery,
+  type SchemaScopedIntrospectionOptions,
+  type SchemaScopedIntrospectionPlan,
+  type SchemaScopedIntrospectionQuery,
+  type SchemaScopedIntrospectionScope,
+  type ScopedCatalogTypes,
+  validateSchemaScopedIntrospection,
+} 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..39532c9deb
--- /dev/null
+++ b/utils/pg-introspection/src/scopedIntrospection.ts
@@ -0,0 +1,552 @@
+import type { Introspection } from "./index.ts";
+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[]];
+}
+
+export interface SchemaScopedIntrospectionScope {
+  schemas: readonly string[];
+  catalogTypes: ScopedCatalogTypes;
+  capabilityExtensions: readonly string[];
+}
+
+export interface SchemaScopedIntrospectionPlan {
+  query: SchemaScopedIntrospectionQuery;
+  scope: SchemaScopedIntrospectionScope;
+}
+
+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 makeSchemaScopedIntrospectionPlan = (
+  schemas: readonly string[],
+  options: SchemaScopedIntrospectionOptions = {},
+): SchemaScopedIntrospectionPlan => {
+  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))";
+  const query: SchemaScopedIntrospectionQuery = {
+    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],
+  };
+  return {
+    query,
+    scope: {
+      schemas: normalized,
+      catalogTypes,
+      capabilityExtensions: normalizedCapabilityExtensions,
+    },
+  };
+};
+
+/**
+ * Builds only the query portion of a schema-scoped introspection plan.
+ *
+ * Prefer `makeSchemaScopedIntrospectionPlan()` when the results will be parsed
+ * and validated by this package.
+ */
+export const makeSchemaScopedIntrospectionQuery = (
+  schemas: readonly string[],
+  options: SchemaScopedIntrospectionOptions = {},
+): SchemaScopedIntrospectionQuery =>
+  makeSchemaScopedIntrospectionPlan(schemas, options).query;
+
+function assertScopedNamespaces(
+  introspection: Introspection,
+  requiredSchemas: readonly string[],
+): void {
+  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 did not find required schema(s): ${missing.join(
+        ", ",
+      )}`,
+    );
+  }
+}
+
+function assertDependencyClosureTypes(introspection: Introspection): void {
+  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 resolves =
+      retainedTypeOids.has(oid) || introspection._lookups.typeById.has(oid);
+    if (!resolves) {
+      throw new Error(
+        `Dependency-closure introspection 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");
+  }
+}
+
+/** Validates that an introspection result satisfies its scoped query plan. */
+export function validateSchemaScopedIntrospection(
+  introspection: Introspection,
+  plan: SchemaScopedIntrospectionPlan,
+): void {
+  assertScopedNamespaces(introspection, plan.scope.schemas);
+  if (plan.scope.catalogTypes === "dependency-closure") {
+    assertDependencyClosureTypes(introspection);
+  }
+}