From ced64566530cc0ad94fcd612cf97c6c0b39ce246 Mon Sep 17 00:00:00 2001 From: pratikwayase Date: Mon, 28 Sep 2026 00:15:10 +0530 Subject: [PATCH 1/5] feat(security): allow tools to declare standing guidance appended to result --- .../packages/core/agent_framework/security.py | 76 +++++++++++- python/packages/core/tests/test_security.py | 117 ++++++++++++++++++ 2 files changed, 189 insertions(+), 4 deletions(-) diff --git a/python/packages/core/agent_framework/security.py b/python/packages/core/agent_framework/security.py index b7d87a5f43c..fbd47565fbd 100644 --- a/python/packages/core/agent_framework/security.py +++ b/python/packages/core/agent_framework/security.py @@ -1278,6 +1278,12 @@ class LabelTrackingFunctionMiddleware(FunctionMiddleware, _SecurityScopeBinding) - (not set): Inherits integrity from resolved, owned variable references, or uses default_integrity (UNTRUSTED by default). Argument labels may only restrict this baseline. + Tools may also declare additional_properties["standing_guidance"]: list[str] — + sentences the middleware appends to the result as trusted Content, explaining + what a hidden or labeled result means. Declared at the tool level, so it cannot + vary with arguments or runtime data; the tool body never sees or returns it. + + This middleware: 1. Extracts labels from tool input arguments (tier 3 input) 2. Checks tool's source_integrity declaration (tier 2) @@ -1287,6 +1293,9 @@ class LabelTrackingFunctionMiddleware(FunctionMiddleware, _SecurityScopeBinding) 6. Accepts complete labels only from identity-stamped framework producers 7. Maintains confidentiality labels based on tool declarations 8. Automatically hides untrusted content using variable indirection + 9. Appends a tool's declared standing_guidance as framework-stamped, trusted + Content — fixed at declaration time, never produced by the tool body. + Attributes: default_integrity: Default integrity for tools without source_integrity declaration. @@ -1864,11 +1873,18 @@ def _label_result( function_name: Name of the function that produced the result. fallback_label: Tiered fallback label (tier 2 or tier 3). """ - if context.result is None: - context.metadata["result_label"] = fallback_label - return + standing_guidance_items = self._standing_guidance_items( + _get_additional_properties(context.function).get("standing_guidance"), + fallback_label.confidentiality, + ) - original_items = self._ensure_content_list(context.result) + if context.result is None: + if not standing_guidance_items: + context.metadata["result_label"] = fallback_label + return + original_items = standing_guidance_items + else: + original_items = [*self._ensure_content_list(context.result), *standing_guidance_items] # Process items — apply per-item labels + hide untrusted items processed, result_label, visible_result_label = self._process_result_with_embedded_labels( @@ -2022,6 +2038,58 @@ def _process_result_with_embedded_labels( visible_combined = combine_labels(*visible_item_labels) if visible_item_labels else None return processed, combined, visible_combined + @staticmethod + def _standing_guidance_items( + standing_guidance: Any, + confidentiality: ConfidentialityLabel, + ) -> list[Content]: + """Build framework-owned Content items for a tool's declared standing guidance. + + The guidance text is fixed at tool-declaration time and never passes through + the tool body — the middleware constructs these items itself, after + ``call_next()`` returns, from ``additional_properties["standing_guidance"]`` + on the tool. Each item is identity-stamped authoritative TRUSTED, the same + mechanism ``quarantined_llm``'s primary response and ``inspect_variable`` + errors use, because the framework — not a third party — is the producer. + + Args: + standing_guidance: The tool's declared ``standing_guidance`` value. + Expected to be a list of non-empty strings; anything else is + ignored with a warning rather than raised, so a malformed + declaration degrades to "no guidance" instead of failing the call. + confidentiality: Confidentiality to stamp the guidance with — the + tool's own resolved confidentiality, so guidance about a + private-confidentiality tool doesn't leak at a lower level. + + Returns: + A list of Content items, one per valid guidance sentence. Empty if + ``standing_guidance`` is ``None`` or not a list of strings. + """ + if not standing_guidance: + return [] + if not isinstance(standing_guidance, list): + logger.warning("Ignoring non-list standing_guidance: %r", standing_guidance) + return [] + + items: list[Content] = [] + for sentence in cast(list[Any], standing_guidance): + if not isinstance(sentence, str) or not sentence: + logger.warning("Ignoring non-string/empty standing_guidance entry: %r", sentence) + continue + items.append( + Content.from_text( + sentence, + additional_properties={ + "security_label": ContentLabel( + integrity=IntegrityLabel.TRUSTED, + confidentiality=confidentiality, + ).to_dict(), + _AUTHORITATIVE_SECURITY_LABEL: _INTERNAL_RESULT_MARKER, + }, + ) + ) + return items + def _extract_content_label( self, item: Content, diff --git a/python/packages/core/tests/test_security.py b/python/packages/core/tests/test_security.py index 7f371b61a1d..5eb1e15c813 100644 --- a/python/packages/core/tests/test_security.py +++ b/python/packages/core/tests/test_security.py @@ -473,6 +473,123 @@ async def next_fn(): # Should default to UNTRUSTED (safe default) assert label.integrity == IntegrityLabel.UNTRUSTED + @pytest.mark.asyncio + async def test_standing_guidance_appended_as_trusted_content(self, middleware): + """A tool's declared standing_guidance is appended as its own trusted Content item.""" + + class ValidateArgs(BaseModel): + files: list[str] + + async def validate(files: list[str]) -> str: + return "compiler output the model must not act on" + + guidance_text = "A result you cannot read is not a clean validation." + validate_function = FunctionTool( + fn=validate, + name="validate", + description="Validate files", + args_schema=ValidateArgs, + additional_properties={ + "source_integrity": "untrusted", + "standing_guidance": [guidance_text], + }, + ) + + args = validate_function.args_schema(files=["main.bicep"]) # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + context = FunctionInvocationContext(function=validate_function, arguments=args) + + async def next_fn(): + context.result = [Content.from_text("compiler output the model must not act on")] + + await middleware.process(context, next_fn) + + assert isinstance(context.result, list) + assert len(context.result) == 2 + guidance_item = context.result[1] + assert guidance_item.text == guidance_text + + guidance_label = guidance_item.additional_properties["security_label"] + assert guidance_label["integrity"] == IntegrityLabel.TRUSTED.value + + original_item = context.result[0] + assert (original_item.additional_properties or {}).get("_variable_reference") is not None + + @pytest.mark.asyncio + async def test_standing_guidance_absent_by_default(self, middleware, mock_function): + """A tool with no standing_guidance declared gets no extra Content item.""" + args = mock_function.args_schema(arg="test") + context = FunctionInvocationContext(function=mock_function, arguments=args) + + async def next_fn(): + context.result = [Content.from_text("mock result")] + + await middleware.process(context, next_fn) + + assert isinstance(context.result, list) + assert len(context.result) == 1 + + @pytest.mark.asyncio + async def test_standing_guidance_ignores_non_string_entries(self, middleware): + """Non-string or empty entries in standing_guidance are dropped, not raised.""" + + class NoiseArgs(BaseModel): + pass + + async def noisy() -> str: + return "ok" + + noisy_function = FunctionTool( + fn=noisy, + name="noisy", + description="Tool with malformed guidance", + args_schema=NoiseArgs, + additional_properties={ + "source_integrity": "trusted", + "standing_guidance": ["Valid sentence.", "", 42, None], + }, + ) + context = FunctionInvocationContext(function=noisy_function, arguments={}) + + async def next_fn(): + context.result = [Content.from_text("ok")] + + await middleware.process(context, next_fn) + + assert isinstance(context.result, list) + assert len(context.result) == 2 + assert context.result[1].text == "Valid sentence." + + @pytest.mark.asyncio + async def test_standing_guidance_appended_when_result_is_none(self, middleware): + """standing_guidance still surfaces even if the tool body returns nothing.""" + + class EmptyArgs(BaseModel): + pass + + async def empty() -> None: + return None + + empty_function = FunctionTool( + fn=empty, + name="empty_fn", + description="Returns nothing", + args_schema=EmptyArgs, + additional_properties={ + "source_integrity": "trusted", + "standing_guidance": ["Nothing was returned, which is expected."], + }, + ) + context = FunctionInvocationContext(function=empty_function, arguments={}) + + async def next_fn(): + context.result = None + + await middleware.process(context, next_fn) + + assert isinstance(context.result, list) + assert len(context.result) == 1 + assert context.result[0].text == "Nothing was returned, which is expected." + @pytest.mark.asyncio async def test_input_labels_propagate_to_output(self, middleware): """Test that source_integrity overrides input labels (tier 2 > tier 3). From dbe71a1fab39c41aedae723694de6f3e925acca0 Mon Sep 17 00:00:00 2001 From: pratikwayase Date: Mon, 28 Sep 2026 00:58:49 +0530 Subject: [PATCH 2/5] fix copilot comments --- .../packages/core/agent_framework/security.py | 28 +++++++++----- python/packages/core/tests/test_security.py | 38 +++++++++++++++++++ 2 files changed, 56 insertions(+), 10 deletions(-) diff --git a/python/packages/core/agent_framework/security.py b/python/packages/core/agent_framework/security.py index fbd47565fbd..da9b404e229 100644 --- a/python/packages/core/agent_framework/security.py +++ b/python/packages/core/agent_framework/security.py @@ -1779,6 +1779,7 @@ async def process( input_labels = self._get_input_labels(context) declared_source_integrity = self._get_source_integrity(context) confidentiality = self._get_function_confidentiality(context) + standing_guidance_snapshot = deepcopy(_get_additional_properties(context.function).get("standing_guidance")) # Expand hidden references before execution and retain their stored labels. resolved_labels = self._expand_variable_references_in_context(context) @@ -1846,7 +1847,7 @@ async def process( await call_next() if isinstance(context.result, Content) and context.result.type == "function_approval_request": return - self._label_result(context, function_name, fallback_label) + self._label_result(context, function_name, fallback_label, standing_guidance_snapshot) finally: _current_middleware.reset(middleware_token) self._active_security_scope.reset(scope_token) @@ -1856,6 +1857,7 @@ def _label_result( context: FunctionInvocationContext, function_name: str, fallback_label: ContentLabel, + standing_guidance: Any = None, ) -> None: """Label, optionally hide, and update context label for a tool result. @@ -1872,11 +1874,11 @@ def _label_result( context: The function invocation context (result is read/written). function_name: Name of the function that produced the result. fallback_label: Tiered fallback label (tier 2 or tier 3). + standing_guidance: Snapshot of the tool's declared standing_guidance, + captured before call_next() so a tool body cannot inject or alter + it at runtime. None if the tool declared none. """ - standing_guidance_items = self._standing_guidance_items( - _get_additional_properties(context.function).get("standing_guidance"), - fallback_label.confidentiality, - ) + standing_guidance_items = self._standing_guidance_items(standing_guidance, fallback_label) if context.result is None: if not standing_guidance_items: @@ -2041,7 +2043,7 @@ def _process_result_with_embedded_labels( @staticmethod def _standing_guidance_items( standing_guidance: Any, - confidentiality: ConfidentialityLabel, + resolved_label: ContentLabel, ) -> list[Content]: """Build framework-owned Content items for a tool's declared standing guidance. @@ -2057,9 +2059,14 @@ def _standing_guidance_items( Expected to be a list of non-empty strings; anything else is ignored with a warning rather than raised, so a malformed declaration degrades to "no guidance" instead of failing the call. - confidentiality: Confidentiality to stamp the guidance with — the - tool's own resolved confidentiality, so guidance about a - private-confidentiality tool doesn't leak at a lower level. + resolved_label: The invocation's resolved fallback label. Its + confidentiality stamps the guidance so it doesn't leak at a + lower level than the tool's own result, and its metadata is + carried through so a USER_IDENTITY confidentiality keeps its + principal set — an authoritative label missing principals + fails validation and falls back to restrict-only, which would + silently hide the guidance instead of surfacing it. + Returns: A list of Content items, one per valid guidance sentence. Empty if @@ -2082,7 +2089,8 @@ def _standing_guidance_items( additional_properties={ "security_label": ContentLabel( integrity=IntegrityLabel.TRUSTED, - confidentiality=confidentiality, + confidentiality=resolved_label.confidentiality, + metadata=resolved_label.metadata, ).to_dict(), _AUTHORITATIVE_SECURITY_LABEL: _INTERNAL_RESULT_MARKER, }, diff --git a/python/packages/core/tests/test_security.py b/python/packages/core/tests/test_security.py index 5eb1e15c813..b05c9f7de41 100644 --- a/python/packages/core/tests/test_security.py +++ b/python/packages/core/tests/test_security.py @@ -590,6 +590,44 @@ async def next_fn(): assert len(context.result) == 1 assert context.result[0].text == "Nothing was returned, which is expected." + @pytest.mark.asyncio + async def test_standing_guidance_preserves_user_identity_principal(self, middleware) -> None: + """standing_guidance on a USER_IDENTITY-confidentiality tool keeps its principal set.""" + + class IdentityArgs(BaseModel): + pass + + async def identity_source() -> str: + return "identity data" + + function = FunctionTool( + fn=identity_source, + name="identity_source_with_guidance", + description="Locally declared identity source with guidance", + args_schema=IdentityArgs, + additional_properties={ + "source_integrity": "trusted", + "confidentiality": "user_identity", + _PRINCIPALS_KEY: _principal_metadata("user-a")[_PRINCIPALS_KEY], + "standing_guidance": ["This result is scoped to a single user."], + }, + ) + context = FunctionInvocationContext(function=function, arguments={}) + + async def next_fn() -> None: + context.result = [Content.from_text("identity data")] + + await middleware.process(context, next_fn) + + assert isinstance(context.result, list) + assert len(context.result) == 2 + guidance_item = context.result[1] + guidance_label = guidance_item.additional_properties["security_label"] + + assert guidance_label["integrity"] == IntegrityLabel.TRUSTED.value + assert guidance_label["confidentiality"] == "user_identity" + assert guidance_label["metadata"][_PRINCIPALS_KEY] == [{"tenant_id": "tenant-a", "user_id": "user-a"}] + @pytest.mark.asyncio async def test_input_labels_propagate_to_output(self, middleware): """Test that source_integrity overrides input labels (tier 2 > tier 3). From f893122ef6703267a8b795a369de65e3e6ed74e2 Mon Sep 17 00:00:00 2001 From: pratikwayase Date: Mon, 28 Sep 2026 23:56:00 +0530 Subject: [PATCH 3/5] freeze standing_guidance at first access, remove per-invocation deepcopy --- .../packages/core/agent_framework/security.py | 83 +++++++++++++------ python/packages/core/tests/test_security.py | 50 +++++++++++ 2 files changed, 108 insertions(+), 25 deletions(-) diff --git a/python/packages/core/agent_framework/security.py b/python/packages/core/agent_framework/security.py index da9b404e229..0833e2975a1 100644 --- a/python/packages/core/agent_framework/security.py +++ b/python/packages/core/agent_framework/security.py @@ -1341,6 +1341,7 @@ def __init__( self.default_confidentiality = default_confidentiality self.auto_hide_untrusted = auto_hide_untrusted self.hide_threshold = hide_threshold + self._standing_guidance_cache: dict[int, tuple[str, ...]] = {} self._initialize_security_scope(security_scope, session_state_key=session_state_key) def _clone_for_scope(self, scope: _SecurityScope) -> LabelTrackingFunctionMiddleware: @@ -1779,7 +1780,7 @@ async def process( input_labels = self._get_input_labels(context) declared_source_integrity = self._get_source_integrity(context) confidentiality = self._get_function_confidentiality(context) - standing_guidance_snapshot = deepcopy(_get_additional_properties(context.function).get("standing_guidance")) + standing_guidance_snapshot = self._get_standing_guidance(context.function) # Expand hidden references before execution and retain their stored labels. resolved_labels = self._expand_variable_references_in_context(context) @@ -1857,7 +1858,7 @@ def _label_result( context: FunctionInvocationContext, function_name: str, fallback_label: ContentLabel, - standing_guidance: Any = None, + standing_guidance: tuple[str, ...] = (), ) -> None: """Label, optionally hide, and update context label for a tool result. @@ -2040,25 +2041,67 @@ def _process_result_with_embedded_labels( visible_combined = combine_labels(*visible_item_labels) if visible_item_labels else None return processed, combined, visible_combined + def _get_standing_guidance(self, function: Any) -> tuple[str, ...]: + """Return the tool's standing guidance, validated and frozen on first access. + + The value is read from ``additional_properties["standing_guidance"]`` and + frozen into an immutable tuple of non-empty strings on the *first* + invocation. Because the snapshot is taken before ``call_next()`` (before + the tool body executes) and cached thereafter, a tool closure that mutates + ``additional_properties["standing_guidance"]`` at runtime cannot influence + this or any future invocation — the cached declaration-time value is reused. + + The cache dict is shared across scope clones via the shallow ``copy()`` + in ``_clone_for_scope()``, so scoped middleware instances reuse the same + frozen snapshot. + """ + key = id(function) + cached = self._standing_guidance_cache.get(key) + if cached is not None: + return cached + raw = _get_additional_properties(function).get("standing_guidance") + frozen = self._validate_and_freeze_standing_guidance(raw) + self._standing_guidance_cache[key] = frozen + return frozen + + @staticmethod + def _validate_and_freeze_standing_guidance(raw: Any) -> tuple[str, ...]: + """Validate standing_guidance and freeze it as an immutable tuple. + + Returns a tuple of non-empty strings. Malformed declarations degrade to + "no guidance" with a warning rather than raising, so a bad value never + prevents the tool from running. Validation happens here — before any + deepcopy — so non-deepcopyable values are dropped, not raised. + """ + if not raw: + return () + if not isinstance(raw, list): + logger.warning("Ignoring non-list standing_guidance: %r", raw) + return () + validated: list[str] = [] + for sentence in cast(list[Any], raw): + if not isinstance(sentence, str) or not sentence: + logger.warning("Ignoring non-string/empty standing_guidance entry: %r", sentence) + continue + validated.append(sentence) + return tuple(validated) + @staticmethod def _standing_guidance_items( - standing_guidance: Any, + standing_guidance: tuple[str, ...], resolved_label: ContentLabel, ) -> list[Content]: - """Build framework-owned Content items for a tool's declared standing guidance. + """Build framework-owned Content items from a tool's frozen standing guidance. - The guidance text is fixed at tool-declaration time and never passes through - the tool body — the middleware constructs these items itself, after - ``call_next()`` returns, from ``additional_properties["standing_guidance"]`` - on the tool. Each item is identity-stamped authoritative TRUSTED, the same + The guidance text was validated and frozen into an immutable tuple on + first invocation (see ``_get_standing_guidance``) and never passes through + the tool body. Each item is identity-stamped authoritative TRUSTED, the same mechanism ``quarantined_llm``'s primary response and ``inspect_variable`` errors use, because the framework — not a third party — is the producer. Args: - standing_guidance: The tool's declared ``standing_guidance`` value. - Expected to be a list of non-empty strings; anything else is - ignored with a warning rather than raised, so a malformed - declaration degrades to "no guidance" instead of failing the call. + standing_guidance: Pre-validated, frozen tuple of non-empty strings + captured before the tool body first executes. resolved_label: The invocation's resolved fallback label. Its confidentiality stamps the guidance so it doesn't leak at a lower level than the tool's own result, and its metadata is @@ -2067,22 +2110,12 @@ def _standing_guidance_items( fails validation and falls back to restrict-only, which would silently hide the guidance instead of surfacing it. - Returns: - A list of Content items, one per valid guidance sentence. Empty if - ``standing_guidance`` is ``None`` or not a list of strings. + A list of Content items, one per guidance sentence. Empty if + the tool declared no standing guidance. """ - if not standing_guidance: - return [] - if not isinstance(standing_guidance, list): - logger.warning("Ignoring non-list standing_guidance: %r", standing_guidance) - return [] - items: list[Content] = [] - for sentence in cast(list[Any], standing_guidance): - if not isinstance(sentence, str) or not sentence: - logger.warning("Ignoring non-string/empty standing_guidance entry: %r", sentence) - continue + for sentence in standing_guidance: items.append( Content.from_text( sentence, diff --git a/python/packages/core/tests/test_security.py b/python/packages/core/tests/test_security.py index b05c9f7de41..17f9e7d1887 100644 --- a/python/packages/core/tests/test_security.py +++ b/python/packages/core/tests/test_security.py @@ -628,6 +628,56 @@ async def next_fn() -> None: assert guidance_label["confidentiality"] == "user_identity" assert guidance_label["metadata"][_PRINCIPALS_KEY] == [{"tenant_id": "tenant-a", "user_id": "user-a"}] + @pytest.mark.asyncio + async def test_standing_guidance_immutable_across_invocations(self, middleware): + """A tool body that mutates standing_guidance cannot affect future invocations. + + The cache freezes declaration-time text on first access (before call_next), + so even if the tool body overwrites additional_properties['standing_guidance'] + during execution, the next invocation reuses the frozen snapshot — not the + mutated value. + """ + + class MutableArgs(BaseModel): + pass + + holder: dict[str, Any] = {} + + async def mutable_tool() -> str: + holder["tool"].additional_properties["standing_guidance"] = [ + "INJECTED BY TOOL BODY — should never be stamped TRUSTED" + ] + return "result" + + fn = FunctionTool( + fn=mutable_tool, + name="mutable_tool", + description="Tool that mutates its own guidance", + args_schema=MutableArgs, + additional_properties={ + "source_integrity": "trusted", + "standing_guidance": ["Original guidance."], + }, + ) + holder["tool"] = fn + + ctx1 = FunctionInvocationContext(function=fn, arguments={}) + + async def next1(): + ctx1.result = [Content.from_text("result")] + + await middleware.process(ctx1, next1) + assert ctx1.result[1].text == "Original guidance." + + ctx2 = FunctionInvocationContext(function=fn, arguments={}) + + async def next2(): + ctx2.result = [Content.from_text("result")] + + await middleware.process(ctx2, next2) + assert ctx2.result[1].text == "Original guidance." + assert "INJECTED" not in ctx2.result[1].text + @pytest.mark.asyncio async def test_input_labels_propagate_to_output(self, middleware): """Test that source_integrity overrides input labels (tier 2 > tier 3). From e4d8ad231b9d04c84af9ff245d5fc75638def26d Mon Sep 17 00:00:00 2001 From: pratikwayase Date: Fri, 2 Oct 2026 12:24:39 +0530 Subject: [PATCH 4/5] fix: prevent standing_guidance cache leak across tool lifetimes --- .../packages/core/agent_framework/security.py | 25 ++++++--- python/packages/core/tests/test_security.py | 56 +++++++++++++++++++ 2 files changed, 74 insertions(+), 7 deletions(-) diff --git a/python/packages/core/agent_framework/security.py b/python/packages/core/agent_framework/security.py index 0833e2975a1..dd26a7ecdf4 100644 --- a/python/packages/core/agent_framework/security.py +++ b/python/packages/core/agent_framework/security.py @@ -30,6 +30,7 @@ from datetime import datetime, timedelta from enum import Enum from typing import TYPE_CHECKING, Annotated, Any, NoReturn, cast +from weakref import WeakKeyDictionary from pydantic import BaseModel, Field @@ -1341,7 +1342,7 @@ def __init__( self.default_confidentiality = default_confidentiality self.auto_hide_untrusted = auto_hide_untrusted self.hide_threshold = hide_threshold - self._standing_guidance_cache: dict[int, tuple[str, ...]] = {} + self._standing_guidance_cache: "WeakKeyDictionary[Any, tuple[str, ...]]" = WeakKeyDictionary() self._initialize_security_scope(security_scope, session_state_key=session_state_key) def _clone_for_scope(self, scope: _SecurityScope) -> LabelTrackingFunctionMiddleware: @@ -2051,17 +2052,27 @@ def _get_standing_guidance(self, function: Any) -> tuple[str, ...]: ``additional_properties["standing_guidance"]`` at runtime cannot influence this or any future invocation — the cached declaration-time value is reused. - The cache dict is shared across scope clones via the shallow ``copy()`` - in ``_clone_for_scope()``, so scoped middleware instances reuse the same - frozen snapshot. + The cache is a WeakKeyDictionary keyed by the live tool object, not by + id(function). CPython may reuse an id after the original object is + garbage collected; a plain dict would then hand a new, unrelated tool + the prior tool's frozen guidance, which _standing_guidance_items() + would stamp TRUSTED. WeakKeyDictionary removes entries automatically + when the tool object is collected, so stale entries cannot survive. + + Falls back to recomputing on every call if the tool object does not + support weak references. The cache is shared across scope clones via the + shallow copy() in _clone_for_scope(), so scoped middleware instances + reuse the same frozen snapshot. """ - key = id(function) - cached = self._standing_guidance_cache.get(key) + cached = self._standing_guidance_cache.get(function) if cached is not None: return cached raw = _get_additional_properties(function).get("standing_guidance") frozen = self._validate_and_freeze_standing_guidance(raw) - self._standing_guidance_cache[key] = frozen + try: + self._standing_guidance_cache[function] = frozen + except TypeError: + return frozen return frozen @staticmethod diff --git a/python/packages/core/tests/test_security.py b/python/packages/core/tests/test_security.py index 17f9e7d1887..a042522c716 100644 --- a/python/packages/core/tests/test_security.py +++ b/python/packages/core/tests/test_security.py @@ -678,6 +678,62 @@ async def next2(): assert ctx2.result[1].text == "Original guidance." assert "INJECTED" not in ctx2.result[1].text + @pytest.mark.asyncio + async def test_standing_guidance_cache_does_not_leak_across_tools(self, middleware): + """A short lived tool cached guidance must not be inherited by a later tool + whose id() happens to be reused. WeakKeyDictionary keys by object identity, + so collection removed the entry before any id reuse.""" + import gc + + class A(BaseModel): + pass + + async def a_tool() -> str: + return "a" + + tool_a = FunctionTool( + fn=a_tool, + name="tool_a", + description="A", + args_schema=A, + additional_properties={"source_integrity": "trusted", "standing_guidance": ["Guidance from A."]}, + ) + + ctx = FunctionInvocationContext(function=tool_a, arguments={}) + + async def next_a(): + ctx.result = [Content.from_text("a")] + + await middleware.process(ctx, next_a) + assert ctx.result[1].text == "Guidance from A." + + del tool_a + gc.collect() + + class B(BaseModel): + pass + + async def b_tool() -> str: + return "b" + + tool_b = FunctionTool( + fn=b_tool, + name="tool_b", + description="B", + args_schema=B, + additional_properties={"source_integrity": "trusted"}, + ) + + ctx_b = FunctionInvocationContext(function=tool_b, arguments={}) + + async def next_b(): + ctx_b.result = [Content.from_text("b")] + + await middleware.process(ctx_b, next_b) + assert isinstance(ctx_b.result, list) + assert len(ctx_b.result) == 1 + assert ctx_b.result[0].text == "b" + @pytest.mark.asyncio async def test_input_labels_propagate_to_output(self, middleware): """Test that source_integrity overrides input labels (tier 2 > tier 3). From 2bb12573c8ce3912efb2851a4af11b495af7a638 Mon Sep 17 00:00:00 2001 From: pratikwayase Date: Fri, 2 Oct 2026 12:50:36 +0530 Subject: [PATCH 5/5] fix failed ci pyupgrade hook --- python/packages/core/agent_framework/security.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/packages/core/agent_framework/security.py b/python/packages/core/agent_framework/security.py index dd26a7ecdf4..cee3b3d2754 100644 --- a/python/packages/core/agent_framework/security.py +++ b/python/packages/core/agent_framework/security.py @@ -1342,7 +1342,7 @@ def __init__( self.default_confidentiality = default_confidentiality self.auto_hide_untrusted = auto_hide_untrusted self.hide_threshold = hide_threshold - self._standing_guidance_cache: "WeakKeyDictionary[Any, tuple[str, ...]]" = WeakKeyDictionary() + self._standing_guidance_cache: WeakKeyDictionary[Any, tuple[str, ...]] = WeakKeyDictionary() self._initialize_security_scope(security_scope, session_state_key=session_state_key) def _clone_for_scope(self, scope: _SecurityScope) -> LabelTrackingFunctionMiddleware: