diff --git a/infra/deploy-runner/build-external-data-plane-artifact.mjs b/infra/deploy-runner/build-external-data-plane-artifact.mjs
index bd13452..0702b69 100644
--- a/infra/deploy-runner/build-external-data-plane-artifact.mjs
+++ b/infra/deploy-runner/build-external-data-plane-artifact.mjs
@@ -96,10 +96,23 @@ async function assertSourceBoundary() {
     'managedReaderBindingProvisioning: config.managedProvisionerApiEnabled',
     'managedReaderBindingLifetime: config.managedProvisionerApiEnabled ? "explicit-revoke" : "disabled"',
     'managedReaderBindings: "digest+idempotent-generation+explicit-revoke"',
+    'readerSourceScope: "writer-resolved+fail-closed-ambiguity"',
+    'source_connection_id as "sourceConnectionId"',
     'where id = $1 and binding_key is null and active = true',
   ]) {
     if (!server.includes(marker)) throw new Error(`managed_reader_boundary_missing:${marker}`);
   }
+  const readerSourceScope = await readFile(
+    resolve(platformRoot, "services/external-data-plane/src/reader-source-scope.mjs"),
+    "utf8",
+  );
+  for (const marker of [
+    "managed_reader_source_scope_not_found",
+    "managed_reader_source_scope_ambiguous",
+    "external_data_plane_writer_bindings",
+  ]) {
+    if (!readerSourceScope.includes(marker)) throw new Error(`reader_source_scope_boundary_missing:${marker}`);
+  }
 }
 
 function canonicalTarScript() {
diff --git a/packages/external-provider-contract/README.md b/packages/external-provider-contract/README.md
index f49122b..93bc06f 100644
--- a/packages/external-provider-contract/README.md
+++ b/packages/external-provider-contract/README.md
@@ -325,6 +325,19 @@ Managed route включается только через
 `EXTERNAL_DATA_PLANE_MANAGED_PROVISIONING_ENABLED`; legacy plaintext routes —
 через отдельный `EXTERNAL_DATA_PLANE_PROVISIONING_ENABLED`.
 
+Managed reader хранит два разных provider-neutral scope. `connectionId` в
+reader grant идентифицирует consumer connection, а `sourceConnectionId`
+фиксирует writer connection, из которого разрешено читать Data Product.
+`sourceConnectionId` не приходит из L2 graph: при ensure EDP разрешает его по
+активным writer bindings того же `tenantId`, `providerId` и полного набора
+`allowedDataProductIds`. Точное совпадение connection выбирается напрямую;
+иначе допускается только один кандидат. Ноль кандидатов возвращает
+`managed_reader_source_scope_not_found`, несколько —
+`managed_reader_source_scope_ambiguous`. Поэтому reader из отдельного L2 target
+не попадает в пустой consumer scope, а несколько provider accounts никогда не
+смешиваются эвристически. Legacy reader binding сохраняет прежнюю семантику:
+его `connectionId` одновременно является source connection.
+
 Managed auth wire contract —
 `nodedc.external-data-plane.managed-provisioner-request/v1`. Engine отправляет
 ровно по одному header:
diff --git a/services/external-data-plane/package.json b/services/external-data-plane/package.json
index 9127d8f..415f018 100644
--- a/services/external-data-plane/package.json
+++ b/services/external-data-plane/package.json
@@ -6,8 +6,8 @@
   "scripts": {
     "start": "node src/server.mjs",
     "dev": "node --watch src/server.mjs",
-    "check": "node --check src/server.mjs && node --check src/schema.mjs && node --check src/config.mjs && node --check src/intake-policy.mjs && node --check src/writer-binding.mjs && node --check src/reader-binding.mjs && node --check src/data-product-policy.mjs && node --check src/data-product-delivery.mjs && node --check src/definitions.mjs && node --check src/managed-provisioner-auth.mjs",
-    "test": "node test/config.test.mjs && node test/managed-provisioner-auth.test.mjs && node test/intake-policy.test.mjs && node test/writer-binding.test.mjs && node test/reader-binding.test.mjs && node test/data-product-policy.test.mjs && node test/definitions.test.mjs",
+    "check": "node --check src/server.mjs && node --check src/schema.mjs && node --check src/config.mjs && node --check src/intake-policy.mjs && node --check src/writer-binding.mjs && node --check src/reader-binding.mjs && node --check src/reader-source-scope.mjs && node --check src/data-product-policy.mjs && node --check src/data-product-delivery.mjs && node --check src/definitions.mjs && node --check src/managed-provisioner-auth.mjs",
+    "test": "node test/config.test.mjs && node test/managed-provisioner-auth.test.mjs && node test/intake-policy.test.mjs && node test/writer-binding.test.mjs && node test/reader-binding.test.mjs && node test/reader-source-scope.test.mjs && node test/data-product-policy.test.mjs && node test/definitions.test.mjs",
     "test:integration": "node test/data-product-delivery.integration.test.mjs && node test/api.integration.test.mjs",
     "test:all": "npm test && npm run test:integration"
   },
diff --git a/services/external-data-plane/src/data-product-delivery.mjs b/services/external-data-plane/src/data-product-delivery.mjs
index 8758953..0c71e5b 100644
--- a/services/external-data-plane/src/data-product-delivery.mjs
+++ b/services/external-data-plane/src/data-product-delivery.mjs
@@ -187,6 +187,7 @@ export async function persistDataProductPublish(pool, batch, definition, {
 }
 
 export async function readDataProductSnapshot(pool, binding, definition, { limit = 5000 } = {}) {
+  const sourceConnectionId = readerSourceConnectionId(binding);
   const client = await pool.connect();
   try {
     await client.query("begin isolation level repeatable read read only");
@@ -202,7 +203,7 @@ export async function readDataProductSnapshot(pool, binding, definition, { limit
        where tenant_id = $1 and connection_id = $2 and provider_id = $3 and data_product_id = $4
        order by source_id asc, semantic_type asc
        limit $5`,
-      [binding.tenantId, binding.connectionId, binding.providerId, definition.id, limit + 1],
+      [binding.tenantId, sourceConnectionId, binding.providerId, definition.id, limit + 1],
     );
     if (rows.rowCount > limit) throw deliveryError("data_product_snapshot_limit_exceeded", 413);
     await client.query("commit");
@@ -266,6 +267,7 @@ export function normalizeHistoryReadOptions(value, definition) {
 }
 
 export async function readDataProductHistory(pool, binding, definition, options) {
+  const sourceConnectionId = readerSourceConnectionId(binding);
   const query = normalizeHistoryReadOptions(options, definition);
   const cursor = query.cursor || {};
   const rows = await pool.query(
@@ -302,7 +304,7 @@ export async function readDataProductHistory(pool, binding, definition, options)
      order by query_bucket asc, source_id asc, semantic_type asc
      limit $13`,
     [
-      binding.tenantId, binding.connectionId, binding.providerId, definition.id,
+      binding.tenantId, sourceConnectionId, binding.providerId, definition.id,
       query.from, query.to, query.sourceIds, query.resolutionMs,
       Boolean(query.cursor), cursor.bucketStart || query.from, cursor.sourceId || "", cursor.semanticType || "",
       query.limit + 1,
@@ -337,6 +339,7 @@ export async function readDataProductHistory(pool, binding, definition, options)
 }
 
 export async function readPatchEvents(pool, binding, definition, after, { limit = 100 } = {}) {
+  const sourceConnectionId = readerSourceConnectionId(binding);
   const client = await pool.connect();
   try {
     await client.query("begin isolation level repeatable read read only");
@@ -351,7 +354,7 @@ export async function readPatchEvents(pool, binding, definition, after, { limit
        where tenant_id = $1 and connection_id = $2 and provider_id = $3
          and data_product_id = $4 and cursor > $5
        order by cursor asc limit $6`,
-      [binding.tenantId, binding.connectionId, binding.providerId, definition.id, after.toString(), limit],
+      [binding.tenantId, sourceConnectionId, binding.providerId, definition.id, after.toString(), limit],
     );
     await client.query("commit");
     return rows.rows.map((row) => ({
@@ -518,26 +521,40 @@ async function persistPatchChunks(client, batch, definition, batchId, chunks) {
 
 async function currentCursor(db, scope, explicitProductId) {
   const dataProductId = explicitProductId || scope.contract.dataProductId;
+  const connectionId = scope.source
+    ? scope.source.connectionId
+    : readerSourceConnectionId(scope);
   const result = await db.query(
     `select current_cursor from external_data_plane_delivery_state
      where tenant_id = $1 and connection_id = $2 and provider_id = $3 and data_product_id = $4`,
-    [scope.tenantId || scope.source.tenantId, scope.connectionId || scope.source.connectionId,
+    [scope.tenantId || scope.source.tenantId, connectionId,
       scope.providerId || scope.source.providerId, dataProductId],
   );
   return result.rowCount ? BigInt(result.rows[0].current_cursor) : 0n;
 }
 
 async function patchFloor(db, binding, dataProductId) {
+  const sourceConnectionId = readerSourceConnectionId(binding);
   const result = await db.query(
     `select min(cursor) as floor from external_data_plane_patch_outbox
      where tenant_id = $1 and connection_id = $2 and provider_id = $3 and data_product_id = $4`,
-    [binding.tenantId, binding.connectionId, binding.providerId, dataProductId],
+    [binding.tenantId, sourceConnectionId, binding.providerId, dataProductId],
   );
   if (result.rows[0]?.floor !== null && result.rows[0]?.floor !== undefined) return BigInt(result.rows[0].floor);
   const current = await currentCursor(db, binding, dataProductId);
   return current + 1n;
 }
 
+function readerSourceConnectionId(binding) {
+  const sourceConnectionId = String(binding?.sourceConnectionId || "").trim();
+  if (sourceConnectionId) return sourceConnectionId;
+  if (!binding?.bindingKey) {
+    const legacyConnectionId = String(binding?.connectionId || "").trim();
+    if (legacyConnectionId) return legacyConnectionId;
+  }
+  throw deliveryError("reader_source_scope_unresolved", 409);
+}
+
 function scopeValues(batch, tail) {
   return [
     batch.source.tenantId, batch.source.connectionId, batch.source.providerId,
diff --git a/services/external-data-plane/src/reader-source-scope.mjs b/services/external-data-plane/src/reader-source-scope.mjs
new file mode 100644
index 0000000..d52758c
--- /dev/null
+++ b/services/external-data-plane/src/reader-source-scope.mjs
@@ -0,0 +1,87 @@
+const IDENTIFIER = /^[a-z][a-z0-9._:-]{2,127}$/;
+
+export async function resolveManagedReaderSourceConnection(db, policy) {
+  const tenantId = identifier(policy?.tenantId);
+  const consumerConnectionId = identifier(policy?.connectionId);
+  const providerId = identifier(policy?.providerId);
+  const allowedDataProductIds = uniqueIdentifiers(policy?.allowedDataProductIds);
+  if (!tenantId || !consumerConnectionId || !providerId || !allowedDataProductIds.length) {
+    throw sourceScopeError("managed_reader_source_scope_request_invalid", 400);
+  }
+
+  const result = await db.query(
+    `select distinct candidate.connection_id as "connectionId"
+     from external_data_plane_writer_bindings as candidate
+     where candidate.tenant_id = $1
+       and candidate.provider_id = $2
+       and candidate.active = true
+       and (candidate.expires_at is null or candidate.expires_at > now())
+       and not exists (
+         select 1
+         from unnest($3::text[]) as requested(data_product_id)
+         where not exists (
+           select 1
+           from external_data_plane_writer_bindings as coverage
+           where coverage.tenant_id = candidate.tenant_id
+             and coverage.connection_id = candidate.connection_id
+             and coverage.provider_id = candidate.provider_id
+             and coverage.active = true
+             and (coverage.expires_at is null or coverage.expires_at > now())
+             and coverage.allowed_data_product_ids ? requested.data_product_id
+         )
+       )
+     order by candidate.connection_id asc
+     limit 3`,
+    [tenantId, providerId, allowedDataProductIds],
+  );
+  const candidates = uniqueIdentifiers(result.rows.map((row) => row.connectionId));
+  if (candidates.includes(consumerConnectionId)) return consumerConnectionId;
+  if (candidates.length === 1) return candidates[0];
+  if (!candidates.length) throw sourceScopeError("managed_reader_source_scope_not_found", 409);
+  throw sourceScopeError("managed_reader_source_scope_ambiguous", 409);
+}
+
+export async function backfillManagedReaderSourceConnections(db) {
+  const unresolved = await db.query(
+    `select id, tenant_id as "tenantId", connection_id as "connectionId",
+       provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds"
+     from external_data_plane_reader_bindings
+     where binding_key is not null and source_connection_id is null
+     order by created_at asc, id asc`,
+  );
+  let resolved = 0;
+  let unresolvedCount = 0;
+  for (const binding of unresolved.rows) {
+    try {
+      const sourceConnectionId = await resolveManagedReaderSourceConnection(db, binding);
+      const result = await db.query(
+        `update external_data_plane_reader_bindings
+         set source_connection_id = $2
+         where id = $1 and binding_key is not null and source_connection_id is null`,
+        [binding.id, sourceConnectionId],
+      );
+      resolved += result.rowCount;
+    } catch (error) {
+      if (!new Set([
+        "managed_reader_source_scope_not_found",
+        "managed_reader_source_scope_ambiguous",
+      ]).has(error?.code)) throw error;
+      unresolvedCount += 1;
+    }
+  }
+  return { resolved, unresolved: unresolvedCount };
+}
+
+function sourceScopeError(code, status) {
+  return Object.assign(new Error(code), { code, status });
+}
+
+function identifier(value) {
+  const normalized = typeof value === "string" ? value.trim() : "";
+  return IDENTIFIER.test(normalized) ? normalized : "";
+}
+
+function uniqueIdentifiers(value) {
+  if (!Array.isArray(value)) return [];
+  return [...new Set(value.map(identifier).filter(Boolean))];
+}
diff --git a/services/external-data-plane/src/schema.mjs b/services/external-data-plane/src/schema.mjs
index 18f1af8..7fff0ac 100644
--- a/services/external-data-plane/src/schema.mjs
+++ b/services/external-data-plane/src/schema.mjs
@@ -1,3 +1,5 @@
+import { backfillManagedReaderSourceConnections } from "./reader-source-scope.mjs";
+
 export async function migrate(pool) {
   await pool.query("create extension if not exists timescaledb");
   await pool.query("create extension if not exists postgis");
@@ -160,6 +162,7 @@ export async function migrate(pool) {
   await pool.query("alter table external_data_plane_reader_bindings add column if not exists binding_key text");
   await pool.query("alter table external_data_plane_reader_bindings add column if not exists request_hash text");
   await pool.query("alter table external_data_plane_reader_bindings add column if not exists generation integer not null default 1");
+  await pool.query("alter table external_data_plane_reader_bindings add column if not exists source_connection_id text");
   await pool.query("alter table external_data_plane_reader_bindings alter column expires_at drop not null");
   await pool.query(`
     do $$
@@ -202,6 +205,13 @@ export async function migrate(pool) {
   `);
   await pool.query("create unique index if not exists external_data_plane_reader_bindings_managed_key_idx on external_data_plane_reader_bindings (binding_key, generation) where binding_key is not null");
   await pool.query("create index if not exists external_data_plane_reader_bindings_active_idx on external_data_plane_reader_bindings (active, expires_at)");
+  await pool.query("create index if not exists external_data_plane_reader_bindings_source_idx on external_data_plane_reader_bindings (tenant_id, source_connection_id, provider_id, active)");
+  await pool.query(`
+    update external_data_plane_reader_bindings
+    set source_connection_id = connection_id
+    where binding_key is null and source_connection_id is null
+  `);
+  await backfillManagedReaderSourceConnections(pool);
 
   await pool.query(`
     create table if not exists external_data_plane_facts (
diff --git a/services/external-data-plane/src/server.mjs b/services/external-data-plane/src/server.mjs
index 0c803c1..c996d9e 100644
--- a/services/external-data-plane/src/server.mjs
+++ b/services/external-data-plane/src/server.mjs
@@ -32,6 +32,7 @@ import {
   readerBindingRequestHash,
   safeReaderBinding,
 } from "./reader-binding.mjs";
+import { resolveManagedReaderSourceConnection } from "./reader-source-scope.mjs";
 import { migrate } from "./schema.mjs";
 import {
   createWriterToken,
@@ -97,6 +98,7 @@ app.get("/healthz", asyncRoute(async (_req, res) => {
     managedWriterBindings: "digest+idempotent-generation+explicit-revoke",
     readerBindings: "supported",
     managedReaderBindings: "digest+idempotent-generation+explicit-revoke",
+    readerSourceScope: "writer-resolved+fail-closed-ambiguity",
     dataProductDelivery: "snapshot+history+durable-patch",
     legacyIntake: config.legacyIntakeEnabled ? "migration-only" : "disabled",
     legacyCapabilityProvisioning: config.provisionerApiEnabled ? "enabled" : "disabled",
@@ -388,7 +390,8 @@ app.put("/internal/data-plane/v1/reader-bindings/by-key/:bindingKey", requireMan
     await client.query("select pg_advisory_xact_lock(hashtextextended($1, 0))", [bindingKey]);
     const existing = await client.query(
       `select id, binding_key as "bindingKey", request_hash as "requestHash", generation,
-         tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
+         tenant_id as "tenantId", connection_id as "connectionId",
+         source_connection_id as "sourceConnectionId", provider_id as "providerId",
          allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
          active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"
        from external_data_plane_reader_bindings
@@ -397,22 +400,40 @@ app.put("/internal/data-plane/v1/reader-bindings/by-key/:bindingKey", requireMan
       [bindingKey, policy.generation],
     );
     if (existing.rowCount) {
-      const binding = existing.rows[0];
+      let binding = existing.rows[0];
       if (binding.requestHash !== requestHash) {
         throw httpError(409, "managed_reader_binding_request_conflict");
       }
       if (binding.active !== true || binding.expiresAt !== null) {
         throw httpError(409, "managed_reader_binding_generation_inactive");
       }
+      if (!binding.sourceConnectionId) {
+        const sourceConnectionId = await resolveManagedReaderSourceConnection(client, policy);
+        const resolved = await client.query(
+          `update external_data_plane_reader_bindings
+           set source_connection_id = $3
+           where binding_key = $1 and generation = $2 and source_connection_id is null
+           returning id, binding_key as "bindingKey", request_hash as "requestHash", generation,
+             tenant_id as "tenantId", connection_id as "connectionId",
+             source_connection_id as "sourceConnectionId", provider_id as "providerId",
+             allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
+             active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
+          [bindingKey, policy.generation, sourceConnectionId],
+        );
+        binding = resolved.rows[0] || binding;
+      }
       response = { status: 200, idempotent: true, binding };
     } else {
+      const sourceConnectionId = await resolveManagedReaderSourceConnection(client, policy);
       const inserted = await client.query(
         `insert into external_data_plane_reader_bindings (
           id, token_hash, binding_key, request_hash, generation,
-          tenant_id, connection_id, provider_id, allowed_data_product_ids, expires_at
-        ) values ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10)
+          tenant_id, connection_id, source_connection_id, provider_id,
+          allowed_data_product_ids, expires_at
+        ) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb, $11)
         returning id, binding_key as "bindingKey", generation,
-          tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
+          tenant_id as "tenantId", connection_id as "connectionId",
+          source_connection_id as "sourceConnectionId", provider_id as "providerId",
           allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
           active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
         [
@@ -423,6 +444,7 @@ app.put("/internal/data-plane/v1/reader-bindings/by-key/:bindingKey", requireMan
           policy.generation,
           policy.tenantId,
           policy.connectionId,
+          sourceConnectionId,
           policy.providerId,
           JSON.stringify(policy.allowedDataProductIds),
           policy.expiresAt,
@@ -459,7 +481,8 @@ app.post("/internal/data-plane/v1/reader-bindings/by-key/:bindingKey/generations
     await client.query("select pg_advisory_xact_lock(hashtextextended($1, 0))", [bindingKey]);
     const existing = await client.query(
       `select id, binding_key as "bindingKey", generation,
-         tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
+         tenant_id as "tenantId", connection_id as "connectionId",
+         source_connection_id as "sourceConnectionId", provider_id as "providerId",
          allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
          active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"
        from external_data_plane_reader_bindings
@@ -474,7 +497,8 @@ app.post("/internal/data-plane/v1/reader-bindings/by-key/:bindingKey/generations
          set active = false, revoked_at = now()
          where binding_key = $1 and generation = $2 and active = true
          returning id, binding_key as "bindingKey", generation,
-           tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
+           tenant_id as "tenantId", connection_id as "connectionId",
+           source_connection_id as "sourceConnectionId", provider_id as "providerId",
            allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
            active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
         [bindingKey, generation],
@@ -505,10 +529,11 @@ app.post("/internal/data-plane/v1/reader-bindings", requireProvisionerApi, async
   const bindingId = randomUUID();
   const result = await pool.query(
     `insert into external_data_plane_reader_bindings (
-      id, token_hash, tenant_id, connection_id, provider_id,
+      id, token_hash, tenant_id, connection_id, source_connection_id, provider_id,
       allowed_data_product_ids, expires_at
-    ) values ($1, $2, $3, $4, $5, $6::jsonb, $7)
+    ) values ($1, $2, $3, $4, $4, $5, $6::jsonb, $7)
     returning id, tenant_id as "tenantId", connection_id as "connectionId",
+      source_connection_id as "sourceConnectionId",
       provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
       expires_at as "expiresAt", active, created_at as "createdAt",
       rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
@@ -533,6 +558,7 @@ app.post("/internal/data-plane/v1/reader-bindings/:bindingId/rotate", requirePro
      set token_hash = $2, rotated_at = now()
      where id = $1 and binding_key is null and active = true and expires_at > now()
      returning id, tenant_id as "tenantId", connection_id as "connectionId",
+       source_connection_id as "sourceConnectionId",
        provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
        expires_at as "expiresAt", active, created_at as "createdAt",
        rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
@@ -549,6 +575,7 @@ app.post("/internal/data-plane/v1/reader-bindings/:bindingId/revoke", requirePro
      set active = false, revoked_at = now()
      where id = $1 and binding_key is null and active = true
      returning id, tenant_id as "tenantId", connection_id as "connectionId",
+       source_connection_id as "sourceConnectionId",
        provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
        expires_at as "expiresAt", active, created_at as "createdAt",
        rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
@@ -937,7 +964,9 @@ async function resolveReaderBinding(req) {
   const token = bearerToken(req);
   if (!token) throw httpError(401, "reader_binding_unauthorized");
   const result = await pool.query(
-    `select id, token_hash as "tokenHash", tenant_id as "tenantId", connection_id as "connectionId",
+    `select id, binding_key as "bindingKey", token_hash as "tokenHash",
+       tenant_id as "tenantId", connection_id as "connectionId",
+       source_connection_id as "sourceConnectionId",
        provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
        expires_at as "expiresAt", active, created_at as "createdAt",
        rotated_at as "rotatedAt", revoked_at as "revokedAt"
@@ -947,6 +976,9 @@ async function resolveReaderBinding(req) {
     [hashReaderToken(token)],
   );
   if (!result.rowCount) throw httpError(401, "reader_binding_unauthorized");
+  if (result.rows[0].bindingKey && !result.rows[0].sourceConnectionId) {
+    throw httpError(409, "reader_source_scope_unresolved");
+  }
   return result.rows[0];
 }
 
diff --git a/services/external-data-plane/test/api.integration.test.mjs b/services/external-data-plane/test/api.integration.test.mjs
index 7eb9fa9..823e068 100644
--- a/services/external-data-plane/test/api.integration.test.mjs
+++ b/services/external-data-plane/test/api.integration.test.mjs
@@ -196,7 +196,7 @@ try {
 
   const managedReaderToken = "ndc_edprb_managed_api_test_0123456789abcdefghijklmnopqrstuvwxyz";
   const managedReaderBody = {
-    source: scope,
+    source: { ...scope, connectionId: "api-consumer-connection" },
     allowedDataProductIds: [productId],
     expiresAt: null,
     generation: 1,
@@ -209,12 +209,25 @@ try {
   ]);
   assert.deepEqual(managedReaderResults.map(({ value }) => value.idempotent).sort(), [false, true]);
   assert.equal(managedReaderResults[0].value.readerBinding.id, managedReaderResults[1].value.readerBinding.id);
+  assert.equal(managedReaderResults[0].value.readerBinding.connectionId, "api-consumer-connection");
   assert.equal("token" in managedReaderResults[0].value, false);
   assert.equal(JSON.stringify(managedReaderResults[0].value).includes(managedReaderBody.capabilityDigest), false);
   assert.deepEqual(
     (await jsonRequest("/internal/data-plane/v1/reader/data-products", { token: managedReaderToken })).dataProducts.map((value) => value.id),
     [productId],
   );
+  const managedSourcePublish = await jsonRequest(`/internal/data-plane/v1/data-products/${productId}/publish`, {
+    method: "POST",
+    token: managedTokenGeneration2,
+    body: publish("api-managed-reader-source", "2026-07-15T11:00:00.000Z", 37.60),
+  });
+  assert.equal(managedSourcePublish.currentUpdatedCount, 1);
+  const managedReaderSnapshot = await jsonRequest(
+    `/internal/data-plane/v1/data-products/${productId}/snapshot`,
+    { token: managedReaderToken },
+  );
+  assert.equal(validateDataProductSnapshot(managedReaderSnapshot).ok, true);
+  assert.equal(managedReaderSnapshot.facts.length, 1);
   const legacyManagedReaderRevoke = await fetch(
     `${baseUrl}/internal/data-plane/v1/reader-bindings/${managedReaderResults[0].value.readerBinding.id}/revoke`,
     { method: "POST", headers: { Authorization: `Bearer ${provisionerSecret}` } },
@@ -236,6 +249,31 @@ try {
   });
   assert.equal(rejectedManagedReader.status, 401);
 
+  const secondWriterToken = "ndc_edpwb_second_source_api_test_0123456789abcdefghijklmnopqrstuvwxyz";
+  await jsonRequest("/internal/data-plane/v1/writer-bindings/by-key/engine.api-connection-2.positions", {
+    method: "PUT",
+    managedSignature: true,
+    body: {
+      ...managedBody,
+      source: { ...scope, connectionId: "api-connection-2" },
+      capabilityDigest: createHash("sha256").update(secondWriterToken, "utf8").digest("hex"),
+    },
+  });
+  const ambiguousReaderToken = "ndc_edprb_ambiguous_api_test_0123456789abcdefghijklmnopqrstuvwxyz";
+  const ambiguousReader = await signedFetch(
+    "/internal/data-plane/v1/reader-bindings/by-key/engine.api-ambiguous.positions-reader",
+    {
+      method: "PUT",
+      body: {
+        ...managedReaderBody,
+        source: { ...scope, connectionId: "api-other-consumer" },
+        capabilityDigest: createHash("sha256").update(ambiguousReaderToken, "utf8").digest("hex"),
+      },
+    },
+  );
+  assert.equal(ambiguousReader.status, 409);
+  assert.equal((await ambiguousReader.json()).error, "managed_reader_source_scope_ambiguous");
+
   const manualWriterBody = { source: scope, allowedDataProductIds: [productId], expiresAt: expiry };
   const signedOnlyManualProvisioning = await signedFetch("/internal/data-plane/v1/writer-bindings", {
     method: "POST",
diff --git a/services/external-data-plane/test/data-product-delivery.integration.test.mjs b/services/external-data-plane/test/data-product-delivery.integration.test.mjs
index efd1f6e..f174754 100644
--- a/services/external-data-plane/test/data-product-delivery.integration.test.mjs
+++ b/services/external-data-plane/test/data-product-delivery.integration.test.mjs
@@ -49,8 +49,10 @@ try {
   const storedDefinition = await loadDataProductDefinition(pool, definition.id);
   const binding = {
     id: "reader-binding-test",
+    bindingKey: "managed-reader-binding-test",
     tenantId: "tenant-test",
-    connectionId: "connection-test",
+    connectionId: "consumer-connection-test",
+    sourceConnectionId: "connection-test",
     providerId: "provider-test",
     allowedDataProductIds: [definition.id],
     active: true,
diff --git a/services/external-data-plane/test/reader-source-scope.test.mjs b/services/external-data-plane/test/reader-source-scope.test.mjs
new file mode 100644
index 0000000..fc74f8e
--- /dev/null
+++ b/services/external-data-plane/test/reader-source-scope.test.mjs
@@ -0,0 +1,42 @@
+import assert from "node:assert/strict";
+import { resolveManagedReaderSourceConnection } from "../src/reader-source-scope.mjs";
+
+const policy = {
+  tenantId: "tenant-01",
+  connectionId: "consumer-connection",
+  providerId: "provider-01",
+  allowedDataProductIds: ["fleet.positions.current.v1"],
+};
+
+assert.equal(
+  await resolveManagedReaderSourceConnection(db(["producer-connection"]), policy),
+  "producer-connection",
+);
+assert.equal(
+  await resolveManagedReaderSourceConnection(db(["producer-connection", "consumer-connection"]), policy),
+  "consumer-connection",
+);
+await assert.rejects(
+  resolveManagedReaderSourceConnection(db([]), policy),
+  (error) => error?.status === 409 && error?.code === "managed_reader_source_scope_not_found",
+);
+await assert.rejects(
+  resolveManagedReaderSourceConnection(db(["producer-a", "producer-b"]), policy),
+  (error) => error?.status === 409 && error?.code === "managed_reader_source_scope_ambiguous",
+);
+await assert.rejects(
+  resolveManagedReaderSourceConnection(db(["producer-connection"]), { ...policy, allowedDataProductIds: [] }),
+  (error) => error?.status === 400 && error?.code === "managed_reader_source_scope_request_invalid",
+);
+
+function db(connectionIds) {
+  return {
+    async query(sql, values) {
+      assert.match(sql, /external_data_plane_writer_bindings/);
+      assert.deepEqual(values, [policy.tenantId, policy.providerId, policy.allowedDataProductIds]);
+      return { rows: connectionIds.map((connectionId) => ({ connectionId })) };
+    },
+  };
+}
+
+console.log("external-data-plane reader source scope: ok");