Preserve installed operator authority through MCP gateway

src/backend/integrations/internal_mcp/gateway.py

@@ -149,9 +149,7 @@ class MethodDefinition:
     # The argument remains hidden from the model and is injected only when the
     # trusted entry point supplies it, but absence does not remove the method
     # from the ordinary-turn capability projection.
-    ordinary_turn_optional_trusted_argument_bindings: (
-        Mapping[str, str] | None
-    ) = None
+    ordinary_turn_optional_trusted_argument_bindings: Mapping[str, str] | None = None
     # Map capability arguments to actor-authorised constrained choices resolved
     # by the trusted entry point. The model may choose only from that bounded
     # set; dispatch maps the stable selector to the current runtime value.
@@ -831,13 +829,21 @@ def _observe_validated_late_completion(
                 user_id, org_id = preexisting_actor_context
             else:
                 user_id, org_id = payload_user_id, payload_org_id
+            # An installed local operator already bound both identity and
+            # provenance. Preserve that explicit authority across this gateway;
+            # ordinary authenticated actors and payload claims do not gain it.
             actor_source = (
-                "preexisting_authenticated_or_workflow_context"
+                INTERNAL_MCP_TRUSTED_LOCAL_OPERATOR_SOURCE
                 if preexisting_actor_context is not None
+                and internal_mcp_actor_context_is_trusted_local_operator()
                 else (
-                    INTERNAL_MCP_TRUSTED_LOCAL_OPERATOR_SOURCE
-                    if self._trusted_actor_payload_fallback
-                    else INTERNAL_MCP_UNTRUSTED_PAYLOAD_ACTOR_SOURCE
+                    "preexisting_authenticated_or_workflow_context"
+                    if preexisting_actor_context is not None
+                    else (
+                        INTERNAL_MCP_TRUSTED_LOCAL_OPERATOR_SOURCE
+                        if self._trusted_actor_payload_fallback
+                        else INTERNAL_MCP_UNTRUSTED_PAYLOAD_ACTOR_SOURCE
+                    )
                 )
             )
             actor_source_token = _ACTOR_CONTEXT_SOURCE.set(actor_source)

tests/backend/test_codex_von_operator_access.py

@@ -113,6 +113,91 @@ def actor():
 
 def test_conflicting_actor_is_rejected_before_canonical_call():
     catalogue = SimpleNamespace(get=lambda name: None)
-    gateway = SimpleNamespace(invoke=lambda *a: pytest.fail("conflicting actor reached handler"))
-    bound = BoundTools({"actor_id": "#V#manager", "organisation_id": "#V#org", "methods": ["task_get"]}, catalogue, gateway)
-    assert bound.call("task_get", {"acting_user_concept_id": "#V#owner"}) == {"success": False, "error_code": "operator_actor_mismatch"}
+    gateway = SimpleNamespace(
+        invoke=lambda *a: pytest.fail("conflicting actor reached handler")
+    )
+    bound = BoundTools(
+        {
+            "actor_id": "#V#manager",
+            "organisation_id": "#V#org",
+            "methods": ["task_get"],
+        },
+        catalogue,
+        gateway,
+    )
+    assert bound.call("task_get", {"acting_user_concept_id": "#V#owner"}) == {
+        "success": False,
+        "error_code": "operator_actor_mismatch",
+    }
+
+
+@pytest.mark.parametrize("operator", [True, False])
+def test_real_gateway_preserves_installed_operator_task_authority(
+    monkeypatch, operator
+):
+    from contextlib import nullcontext
+    from src.backend.integrations.internal_mcp.catalogue import (
+        _authorise_task_actor_mutation,
+    )
+    from src.backend.integrations.internal_mcp.gateway import (
+        InternalMCPGateway,
+        MethodCatalogue,
+        MethodDefinition,
+        INTERNAL_MCP_TRUSTED_LOCAL_OPERATOR_SOURCE,
+        bind_internal_mcp_actor_context_source,
+    )
+    from src.backend.integrations.internal_mcp.schemas import Schema
+    from src.backend.integrations.internal_mcp.transport import InternalMCPTransport
+    from src.backend.security.access_control import override_current_actor
+    from src.backend.services import task_management_service
+
+    task = {
+        "task_concept_id": "#V#task",
+        "organisation_concept_id": "#V#org",
+        "assignee_concept_id": "#V#other_agent",
+        "created_by_concept_id": "#V#owner",
+    }
+    monkeypatch.setattr(task_management_service, "get_task", lambda _: task)
+
+    def handler(**kwargs):
+        actor, found, error = _authorise_task_actor_mutation(
+            kwargs, surface="task_add_comment"
+        )
+        return error or {
+            "success": True,
+            "actor": actor.user_concept_id,
+            "task": found["task_concept_id"],
+        }
+
+    catalogue = MethodCatalogue()
+    catalogue.register(
+        MethodDefinition(
+            name="task_authority_probe",
+            handler=handler,
+            input_schema=Schema(required={}, optional={}, allow_unknown=True),
+        )
+    )
+    gateway = InternalMCPGateway(
+        catalogue=catalogue, transport=InternalMCPTransport(), enabled=True
+    )
+    provenance = (
+        bind_internal_mcp_actor_context_source(
+            INTERNAL_MCP_TRUSTED_LOCAL_OPERATOR_SOURCE,
+            preexisting_actor_context=("#V#manager", "#V#org"),
+        )
+        if operator
+        else nullcontext()
+    )
+    with override_current_actor("#V#manager", "#V#org"), provenance:
+        result = gateway.invoke(
+            "task_authority_probe",
+            {
+                "task_concept_id": "#V#task",
+                "actor_context_source": INTERNAL_MCP_TRUSTED_LOCAL_OPERATOR_SOURCE,
+            },
+        ).payload
+    if operator:
+        assert result == {"success": True, "actor": "#V#manager", "task": "#V#task"}
+    else:
+        assert result["success"] is False
+        assert result["error_code"] == "task_actor_scope_denied"