Skip to content
Open
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 15 additions & 8 deletions python/semantic_kernel/functions/kernel_arguments.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,10 +67,15 @@ def __or__(self, value: dict) -> "KernelArguments":
f"TypeError: unsupported operand type(s) for |: '{type(self).__name__}' and '{type(value).__name__}'"
)

# Merge execution settings
new_execution_settings = (self.execution_settings or {}).copy()
# Merge execution settings - only copy when needed
if self.execution_settings:
new_execution_settings = self.execution_settings # Reuse reference if immutable
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Aliasing bug: when this branch is taken (self.execution_settings is truthy) and the if isinstance(value, KernelArguments) branch below is NOT taken, new_execution_settings is the same object as self.execution_settings. The constructor at line 46-47 assigns a dict settings argument directly without copying, so result.execution_settings is self.execution_settings. Mutating one will silently corrupt the other. The original .copy() was correct and should be kept.

Suggested change
# Merge execution settings - only copy when needed
if self.execution_settings:
new_execution_settings = self.execution_settings # Reuse reference if immutable
# Merge execution settings
new_execution_settings = (self.execution_settings or {}).copy()
if isinstance(value, KernelArguments) and value.execution_settings:
new_execution_settings |= value.execution_settings

else:
new_execution_settings = {}

if isinstance(value, KernelArguments) and value.execution_settings:
new_execution_settings |= value.execution_settings
# Only copy when we need to merge (mutation)
new_execution_settings = {**new_execution_settings, **value.execution_settings}
# Create a new KernelArguments with merged dict values
return KernelArguments(settings=new_execution_settings, **(dict(self) | dict(value)))

Expand All @@ -84,12 +89,13 @@ def __ror__(self, value: dict) -> "KernelArguments":
f"TypeError: unsupported operand type(s) for |: '{type(value).__name__}' and '{type(self).__name__}'"
)

# Merge execution settings
# Merge execution settings - only copy when needed
new_execution_settings = {}
if isinstance(value, KernelArguments) and value.execution_settings:
new_execution_settings = value.execution_settings.copy()
new_execution_settings = value.execution_settings # Reuse reference if immutable
if self.execution_settings:
new_execution_settings |= self.execution_settings
# Only copy when we need to merge (mutation)
new_execution_settings = {**new_execution_settings, **self.execution_settings}

# Create a new KernelArguments with merged dict values
return KernelArguments(settings=new_execution_settings, **(dict(value) | dict(self)))
Expand All @@ -98,12 +104,13 @@ def __ior__(self, value: "SupportsKeysAndGetItem[Any, Any] | Iterable[tuple[Any,
"""Merges into this KernelArguments with another KernelArguments or dict (in-place)."""
self.update(value)

# In-place merge execution settings
# In-place merge execution settings - only copy when needed
if isinstance(value, KernelArguments) and value.execution_settings:
if self.execution_settings:
self.execution_settings.update(value.execution_settings)
else:
self.execution_settings = value.execution_settings.copy()
# Only copy when we actually need a different instance
self.execution_settings = {**value.execution_settings}

return self

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,121 @@
# Copyright (c) Microsoft. All rights reserved.

"""Tests for KernelArguments merge optimization (Issue #2: Lazy dict copy)."""

import pytest
from unittest.mock import patch, MagicMock

from semantic_kernel.functions.kernel_arguments import KernelArguments


class TestKernelArgumentsMergeOptimization:
"""Test suite for KernelArguments merge operator optimization."""

def test_or_operator_no_execution_settings_copy(self):
"""Test that | operator doesn't copy when execution_settings is None or empty.

This tests the optimization where dict copy is avoided when not needed.
"""
args = KernelArguments(a=1, b=2)
args.execution_settings = None

result = args | {"c": 3}

# Result should have merged values
assert result["a"] == 1
assert result["b"] == 2
assert result["c"] == 3

def test_or_operator_with_kernel_arguments_merge(self):
"""Test | operator with another KernelArguments with execution_settings."""
settings1 = {"model": "gpt-4", "temperature": 0.7}
settings2 = {"temperature": 0.9, "max_tokens": 100}

args1 = KernelArguments(a=1, settings=settings1)
args2 = KernelArguments(b=2, settings=settings2)

result = args1 | args2

# Check merged arguments
assert result["a"] == 1
assert result["b"] == 2

# Check merged execution settings
assert result.execution_settings["model"] == "gpt-4"
assert result.execution_settings["temperature"] == 0.9 # args2 overwrites
assert result.execution_settings["max_tokens"] == 100
Comment on lines +26 to +41
Copy link

Copilot AI Mar 24, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The tests construct KernelArguments(settings=...) with plain dicts like {"model": "gpt-4"} and then assert on keys like temperature. In the implementation, execution_settings is documented/typed as dict[str, PromptExecutionSettings] keyed by service_id, so these tests are not exercising the real semantics and may be misleading (and could hide regressions around service-id overwrites). Prefer using PromptExecutionSettings(service_id=...) instances (or a dict of them) so the tests match actual usage.

Copilot uses AI. Check for mistakes.

def test_ror_operator_lazy_copy(self):
"""Test reverse | operator avoids copy when execution_settings is empty."""
args = KernelArguments(a=1, b=2)
args.execution_settings = {}

result = {"c": 3} | args

# Result should have merged values
assert result["a"] == 1
assert result["b"] == 2
assert result["c"] == 3

def test_ior_operator_lazy_copy(self):
"""Test in-place |= operator avoids copy when execution_settings exists."""
settings1 = {"model": "gpt-4", "temperature": 0.7}
args1 = KernelArguments(a=1, settings=settings1)

settings2 = {"temperature": 0.9}
args2 = KernelArguments(b=2, settings=settings2)

args1 |= args2

# Check merged in-place
assert args1["a"] == 1
assert args1["b"] == 2

# Check in-place merge of settings
assert args1.execution_settings["model"] == "gpt-4"
assert args1.execution_settings["temperature"] == 0.9

def test_ior_operator_creates_copy_when_needed(self):
"""Test that |= creates new dict only when target has no execution_settings."""
args1 = KernelArguments(a=1)
args1.execution_settings = None

settings2 = {"model": "gpt-4"}
args2 = KernelArguments(b=2, settings=settings2)

args1 |= args2

# Should have new execution_settings dict
assert args1.execution_settings is not None
assert args1.execution_settings["model"] == "gpt-4"
# Should not be the same reference (it's a copy)
assert args1.execution_settings is not args2.execution_settings

def test_or_operator_preserves_original_settings(self):
"""Test that | operator doesn't mutate original execution_settings."""
original_settings = {"model": "gpt-4", "temperature": 0.7}
args1 = KernelArguments(a=1, settings=original_settings.copy())
args2 = KernelArguments(b=2, settings={"temperature": 0.9})

result = args1 | args2
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This test only covers the case where both sides have execution_settings (which correctly creates a new dict via spread). The dangerous untested path is when only the LHS has settings and the RHS is a plain dict — where the shared-reference bug in __or__ manifests. Add a test like:

def test_or_operator_with_plain_dict_preserves_original_settings(self):
    args1 = KernelArguments(a=1, settings=PromptExecutionSettings(service_id='s1'))
    result = args1 | {'b': 2}
    assert result.execution_settings is not args1.execution_settings


# Original args1 settings should be unchanged
assert args1.execution_settings["temperature"] == 0.7
# Result should have merged settings
assert result.execution_settings["temperature"] == 0.9

def test_ior_operator_merges_into_existing_dict(self):
"""Test that |= merges into existing settings dict when present."""
settings1 = {"model": "gpt-4", "temperature": 0.7}
args1 = KernelArguments(a=1, settings=settings1)
original_dict = args1.execution_settings

settings2 = {"temperature": 0.9, "max_tokens": 100}
args2 = KernelArguments(b=2, settings=settings2)

args1 |= args2

# Should have reused and updated the original dict
assert args1.execution_settings is original_dict
assert args1.execution_settings["temperature"] == 0.9
assert args1.execution_settings["max_tokens"] == 100