diff --git a/loopx/control_plane/collaboration/source_grant_observation.py b/loopx/control_plane/collaboration/source_grant_observation.py
index 48ed930282..145b92dd43 100644
--- a/loopx/control_plane/collaboration/source_grant_observation.py
+++ b/loopx/control_plane/collaboration/source_grant_observation.py
@@ -1,7 +1,7 @@
 """Provider/store observation for the typed context-recipient policy owner."""
 
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 
 from ...agent_registry import registered_agent_ids_for_goal
 from ..goals.activation import goal_is_stopped
@@ -18,10 +18,21 @@
 POLICY_SCHEMA = "loopx_manager_context_policy_v1"
 
 
+def _source_ingress(runtime_root: Path, session: dict[str, Any], turn: dict[str, Any]) -> dict[str, Any]:
+    return _read(_root(runtime_root) / "ingress" /
+                 (_hash([session["session_id"], turn["client_turn_id"]]) + ".json"))
+
+
+def _source_policy(runtime_root: Path, ingress: dict[str, Any]) -> dict[str, Any]:
+    policy = _read(_root(runtime_root) / "policy.json")
+    if policy.get("schema_version") != POLICY_SCHEMA:
+        raise ValueError("invalid policy")
+    return cast(dict[str, Any], policy.get("sources", {}).get(ingress["channel"], {}))
+
+
 def external_source_policy(runtime_root: Path, session: dict[str, Any], turn: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
     """Observe host provenance for the shared typed source-policy owner."""
-    ingress = _read(_root(runtime_root) / "ingress" /
-                    (_hash([session["session_id"], turn["client_turn_id"]]) + ".json"))
+    ingress = _source_ingress(runtime_root, session, turn)
     try:
         effect_runtime_result("collaboration.source.ingress", {
             "session": {key: session.get(key) for key in ("session_id", "goal_id", "channel_id", "steward_context")},
@@ -33,16 +44,13 @@ def external_source_policy(runtime_root: Path, session: dict[str, Any], turn: di
     except EffectRuntimeRejected as exc:
         raise ValueError(str(exc)) from exc
 
-    policy = _read(_root(runtime_root) / "policy.json")
-    if policy.get("schema_version") != POLICY_SCHEMA:
-        raise ValueError("invalid policy")
-    return ingress, policy.get("sources", {}).get(ingress["channel"], {})
+    return ingress, _source_policy(runtime_root, ingress)
 
 
-def registered_context_recipients(registry: dict) -> dict:
+def registered_context_recipients(registry: dict[str, Any]) -> dict[str, Any]:
     """Observe active Goal membership, ignoring unreadable activation rows."""
     active_goals = []
-    available = []
+    available: list[dict[str, Any]] = []
     for goal in registry.get("goals", []):
         if not isinstance(goal, dict) or not goal.get("id"):
             continue
@@ -59,69 +67,54 @@ def registered_context_recipients(registry: dict) -> dict:
 
 def _source_context_grant(
     runtime_root: Path,
-    session: dict,
-    turn: dict,
-    available_rows: list[dict],
-) -> dict:
-    available = {
-        (row["goal_id"], row["agent_id"])
-        for row in available_rows
-        if isinstance(row, dict)
-        and isinstance(row.get("goal_id"), str)
-        and isinstance(row.get("agent_id"), str)
-    }
-    scope = conversation_scope(session, origin=turn.get("origin", "unknown"))
-    if scope["private_conversation"] and turn.get("origin") == "web":
-        allowed = {
-            target
-            for target in available
-            if scope["goal_ids"] is None or target[0] in scope["goal_ids"]
-        }
-        source_id = "web:" + _hash([session["session_id"], turn["client_turn_id"]])
-    else:
-        if scope["kind"] != "external_audience":
-            return {"mode": "unavailable", "targets": []}
+    session: dict[str, Any],
+    turn: dict[str, Any],
+    available_rows: list[dict[str, Any]],
+) -> dict[str, Any]:
+    # The channel selects stored ingress IO, never delivery authority. Core
+    # classifies the canonical conversation and decides the complete grant.
+    source = None
+    ingress = None
+    source_id = None
+    channel = session.get("channel_id")
+    if isinstance(channel, str) and channel.startswith("manager.external."):
         try:
-            ingress, grants = external_source_policy(runtime_root, session, turn)
-            selected = effect_runtime_result(
-                "collaboration.source.recipients",
-                {
-                    "source": grants,
-                    "sender_id": ingress["sender_id"],
-                    "available": available_rows,
-                },
-            )
-            allowed = {
-                (value["goal_id"], value["agent_id"])
-                for value in selected["targets"]
-            }
-            source_id = ingress["source_id"]
+            ingress = _source_ingress(runtime_root, session, turn)
+            source = _source_policy(runtime_root, ingress)
         except (
             OSError,
             ValueError,
             KeyError,
             TypeError,
             AttributeError,
-            EffectRuntimeRejected,
         ):
-            return {"mode": "unavailable", "targets": []}
-    targets = [
-        {"goal_id": goal_id, "agent_id": agent_id}
-        for goal_id, agent_id in sorted(allowed & available)
-    ]
-    return {
-        "mode": "context_only",
-        "targets": targets,
-        "source_id": source_id,
-    }
+            source = None
+    elif turn.get("origin") == "web" and "session_id" in session and "client_turn_id" in turn:
+        source_id = "web:" + _hash([session["session_id"], turn["client_turn_id"]])
+    return cast(dict[str, Any], effect_runtime_result(
+        "collaboration.source.context_authority",
+        {
+            "conversation": {key: session.get(key) for key in (
+                "session_id", "goal_id", "channel_id", "project_context", "steward_context",
+            )} | {"origin": turn.get("origin", "unknown")},
+            "turn": {"client_turn_id": turn.get("client_turn_id"), "origin": turn.get("origin"),
+                     "message_digest": _hash(turn.get("message"))} if ingress is not None else None,
+            "ingress": {key: ingress[key] for key in (
+                "session_id", "client_turn_id", "channel", "message_digest", "sender_id", "source_id", "origin",
+            ) if key in ingress} if ingress is not None else None,
+            "available": available_rows,
+            "source": source,
+            "source_id": source_id,
+        },
+    ))
 
 
 def source_context_target_authority(
     runtime_root: Path,
-    session: dict,
-    turn: dict,
-    target: dict,
-) -> dict:
+    session: dict[str, Any],
+    turn: dict[str, Any],
+    target: dict[str, Any],
+) -> dict[str, Any]:
     """Authorize one target whose exact Goal scope was already validated."""
     return _source_context_grant(runtime_root, session, turn, [target])
 
@@ -177,8 +170,8 @@ def source_execution_bindings(
 
 
 def source_context_authority(
-    runtime_root: Path, registry_path: Path, session: dict, turn: dict
-) -> dict:
+    runtime_root: Path, registry_path: Path, session: dict[str, Any], turn: dict[str, Any]
+) -> dict[str, Any]:
     """Return only a write-only recipient catalog; no cross-audience Goal evidence."""
     if registry_path is None:
         return {"mode": "unavailable", "targets": []}
diff --git a/loopx/control_plane/collaboration/source_grants.ts b/loopx/control_plane/collaboration/source_grants.ts
index da3e690837..8a815a23de 100644
--- a/loopx/control_plane/collaboration/source_grants.ts
+++ b/loopx/control_plane/collaboration/source_grants.ts
@@ -1,6 +1,7 @@
 import type { JsonObject } from "../effect_program.ts";
 import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
 import { requireBoolean, requireJsonObject, requireNonEmptyString, requireStringArray } from "../runtime_decode.ts";
+import { authorityUnicodeCompare } from "../coordination/authority_store_codec.ts";
 import { normalizeStewardGoalScope, resolveConversationScope } from "./conversation_scope.ts";
 
 type Recipient = { goal_id: string; agent_id?: string };
@@ -101,6 +102,47 @@ export function resolveSourceRecipients(params: JsonObject): JsonObject {
     a.goal_id.localeCompare(b.goal_id) || a.agent_id!.localeCompare(b.agent_id!)) };
 }
 
+/** Combine the canonical conversation scope with observed recipient grants.
+ * Python supplies registry/policy IO and the original source identity. Neither a
+ * project workspace grant nor an unverified source becomes delivery authority.
+ */
+export function resolveSourceContextAuthority(params: JsonObject): JsonObject {
+  const conversation = requireJsonObject(params.conversation, "source conversation");
+  const scope = resolveConversationScope(conversation);
+  if (!Array.isArray(params.available)) throw new EffectRuntimeRequestError("registered recipients unavailable");
+  let selected: Required<Recipient>[];
+  let sourceId = params.source_id;
+  if (scope.private_conversation && conversation.origin === "web") {
+    // Match the registry observation's historical treatment of unreadable rows.
+    const available = params.available.filter((raw): raw is JsonObject & Required<Recipient> =>
+      raw !== null && typeof raw === "object" && !Array.isArray(raw)
+      && typeof raw.goal_id === "string" && typeof raw.agent_id === "string");
+    selected = available.filter(row => scope.goal_ids === null
+      || scope.goal_ids.some(goal => goal === row.goal_id));
+  } else {
+    if (scope.kind !== "external_audience" || params.source === null) {
+      return {mode: "unavailable", targets: []};
+    }
+    try {
+      validateSourceIngress({session: conversation, turn: params.turn, ingress: params.ingress});
+      const ingress = requireJsonObject(params.ingress, "source ingress");
+      selected = resolveSourceRecipients({source: params.source,
+        sender_id: ingress.sender_id, available: params.available}).targets as Required<Recipient>[];
+      sourceId = ingress.source_id;
+    } catch (error) {
+      if (!(error instanceof EffectRuntimeRequestError)) throw error;
+      return {mode: "unavailable", targets: []};
+    }
+  }
+  const targets = new Map(selected.map(row => [JSON.stringify([row.goal_id, row.agent_id]),
+    {goal_id: row.goal_id, agent_id: row.agent_id}]));
+  // The combined authority catalog has always used code-point ordering, even
+  // though the underlying source-policy view uses locale ordering.
+  return {mode: "context_only", source_id: requireNonEmptyString(sourceId, "original source identity"),
+    targets: [...targets.values()].sort((a, b) => authorityUnicodeCompare(a.goal_id, b.goal_id)
+      || authorityUnicodeCompare(a.agent_id, b.agent_id))};
+}
+
 /** Verify the host's immutable ingress against its canonical Session/Turn.
  * IO supplies digests, not a second provenance rule or model-authored identity.
  * Local continuation retains the binding's narrow audience and source policy.
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index ab48e0118a..953d5e419b 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -623,6 +623,7 @@ export function createEffectRuntimeHandlers(
     ["collaboration.conversation.agent_target", lazyHandler(() => import("./collaboration/conversation_binding.ts"), ({resolveConversationAgentTarget}) => resolveConversationAgentTarget)],
     ["collaboration.peer.context_access", lazyHandler(() => import("./collaboration/peer_context.ts"), ({requirePeerContextAccess}) => requirePeerContextAccess)],
     ["collaboration.source.recipients", lazyHandler(() => import("./collaboration/source_grants.ts"), ({resolveSourceRecipients}) => resolveSourceRecipients)],
+    ["collaboration.source.context_authority", lazyHandler(() => import("./collaboration/source_grants.ts"), ({resolveSourceContextAuthority}) => resolveSourceContextAuthority)],
     ["collaboration.source.ingress", lazyHandler(() => import("./collaboration/source_grants.ts"), ({validateSourceIngress}) => validateSourceIngress)],
     ["collaboration.source.execution_bindings", lazyHandler(() => import("./collaboration/source_grants.ts"), ({sourceExecutionBindings}) => sourceExecutionBindings)],
     ["collaboration.source.configure_recipient", lazyHandler(() => import("./collaboration/source_grants.ts"), ({configureSourceRecipient}) => configureSourceRecipient)],
diff --git a/tests/control_plane_ts/source_grants.test.ts b/tests/control_plane_ts/source_grants.test.ts
index df102be292..a6a4249fdd 100644
--- a/tests/control_plane_ts/source_grants.test.ts
+++ b/tests/control_plane_ts/source_grants.test.ts
@@ -1,6 +1,6 @@
 import assert from "node:assert/strict";
 import test from "node:test";
-import { configureSourceRecipient, configureSourceScope, resolveSourceRecipients, sourceExecutionBindings, validateSourceIngress } from "../../loopx/control_plane/collaboration/source_grants.ts";
+import { configureSourceRecipient, configureSourceScope, resolveSourceContextAuthority, resolveSourceRecipients, sourceExecutionBindings, validateSourceIngress } from "../../loopx/control_plane/collaboration/source_grants.ts";
 
 const worker = { goal_id: "research", agent_id: "worker" };
 const peer = { goal_id: "research", agent_id: "peer" };
@@ -8,6 +8,55 @@ const other = { goal_id: "other", agent_id: "worker" };
 const available = [worker, peer, other];
 const source = { local_delivery_scope: "selected", sender_ids: ["owner"], targets: [{ goal_id: "research" }] };
 
+test("source authority keeps local portfolio and same-Goal delivery distinct", () => {
+  const params = {conversation: {channel_id: "manager", origin: "web"},
+    available: [worker, peer, other, worker], source_id: "web:original"};
+  assert.deepEqual(resolveSourceContextAuthority(params),
+    {mode: "context_only", source_id: params.source_id, targets: [other, peer, worker]});
+  assert.deepEqual(resolveSourceContextAuthority({...params,
+    conversation: {channel_id: "goal.research", goal_id: "research", origin: "web"}}).targets,
+    [peer, worker]);
+  // A read-only workspace is not a Goal or a portfolio delivery grant.
+  const project = {kind: "project_workspace", project_ref: "a".repeat(24),
+    workspace_path: "/synthetic/project", audience: "local_owner", grant: "workspace_read"};
+  assert.deepEqual(resolveSourceContextAuthority({...params, conversation: {
+    channel_id: `project.${project.project_ref}`, goal_id: null, project_context: project, origin: "web"}}),
+    {mode: "context_only", source_id: params.source_id, targets: []});
+  assert.throws(() => resolveSourceContextAuthority({...params, source_id: ""}));
+});
+
+test("source authority preserves exact external grants, revocations and original identity", () => {
+  const conversation = {session_id: "original", channel_id: "manager.external.synthetic", origin: "lark"};
+  const turn = {client_turn_id: "input", origin: "lark", message_digest: "digest"};
+  const ingress = {session_id: "original", client_turn_id: "input", channel: conversation.channel_id,
+    message_digest: "digest", sender_id: "owner", source_id: "provider:original"};
+  const params = {conversation, turn, ingress, available, source: {...source, blocked_targets: [peer]},
+    source_id: "not-a-substitute-for-ingress"};
+  assert.deepEqual(resolveSourceContextAuthority(params),
+    {mode: "context_only", source_id: ingress.source_id, targets: [worker]});
+  const unavailable = {mode: "unavailable", targets: []};
+  for (const change of [{source: null}, {ingress: {...ingress, sender_id: "visitor"}},
+    {ingress: {...ingress, session_id: "another"}}, {turn: {...turn, message_digest: "changed"}},
+    {ingress: null},
+    {source: {...source, blocked_targets: null}}, {available: [worker, {}]}]) {
+    assert.deepEqual(resolveSourceContextAuthority({...params, ...change}), unavailable);
+  }
+  assert.deepEqual(resolveSourceContextAuthority({...params,
+    conversation: {channel_id: "unavailable", origin: "lark"}}), unavailable);
+  assert.deepEqual(resolveSourceContextAuthority({...params,
+    conversation: {channel_id: "goal.research", goal_id: "research", origin: "lark"}}), unavailable);
+});
+
+test("source authority has one canonical deduplicated catalog for observed identities", () => {
+  const rows = [{goal_id: "research", agent_id: "a"}, {goal_id: "research", agent_id: "A"},
+    {goal_id: "research", agent_id: "\u{10000}"}, {goal_id: "research", agent_id: "\uE000"}];
+  const params = {conversation: {channel_id: "manager", origin: "web"}, source_id: "web:original",
+    available: [...rows, rows[0], null, {}]};
+  assert.deepEqual(resolveSourceContextAuthority(params).targets, [rows[1], rows[0], rows[3], rows[2]]);
+  assert.deepEqual(resolveSourceContextAuthority({...params, available: []}),
+    {mode: "context_only", source_id: params.source_id, targets: []});
+});
+
 test("execution needs an independent exact grant, current sender and both registered identities", () => {
   const binding = {...worker, binding_id: "review", requester_agent_id: "peer"};
   const params = {source, sender_id: "owner", available};