From 34377737186d51d681eb66c48aabb09801cc8367 Mon Sep 17 00:00:00 2001
From: finnoybu <21694570+finnoybu@users.noreply.github.com>
Date: Mon, 18 May 2026 06:14:34 -0400
Subject: [PATCH 1/4] Python: feat: cross-session origin attribution on context
 messages

Add an optional origin_session_id parameter to SessionContext.extend_messages
that propagates into the existing _attribution payload on
Message.additional_properties. Downstream context observers can use it to
detect when a provider injects content stored under a different session than
the requesting one.

Populate the field from the harness memory consolidation pipeline
(_harness/_memory.py) when injected topics include contributions from
sessions other than the current one. Add a self-contained sample observer
under samples/02-agents/context_providers/cross_session_observer.py
demonstrating how to subscribe to the signal.

Backward-compatible: omitting the parameter preserves the existing
attribution shape exactly. Tests added in test_sessions.py and
test_harness_memory.py cover the new parameter, the harness cross-session
case, and the same-session case.

Motivated by Dai et al., Stateful Agent Backdoor (arXiv:2605.06158, May
2026), which specifically surveys MAF in section 6.1 / Table 10.
See #5914 for design discussion.

Surfaced during independent audit conducted by @finnoybu (Ken Tannenbaum, AEGIS Initiative); [MEDIUM, python/packages/core].
---
 .../core/agent_framework/_harness/_memory.py  |  21 +++
 .../core/agent_framework/_sessions.py         |  20 ++-
 .../core/tests/core/test_harness_memory.py    |  88 +++++++++++
 .../packages/core/tests/core/test_sessions.py |  33 ++++
 .../02-agents/context_providers/README.md     |   4 +
 .../cross_session_observer.py                 | 141 ++++++++++++++++++
 6 files changed, 306 insertions(+), 1 deletion(-)
 create mode 100644 python/samples/02-agents/context_providers/cross_session_observer.py

diff --git a/python/packages/core/agent_framework/_harness/_memory.py b/python/packages/core/agent_framework/_harness/_memory.py
index 92e060f4428..9fdb9149173 100644
--- a/python/packages/core/agent_framework/_harness/_memory.py
+++ b/python/packages/core/agent_framework/_harness/_memory.py
@@ -1306,6 +1306,26 @@ async def consolidate_memories() -> str:
         )
         if recent_history_messages:
             context.extend_messages(self.source_id, recent_history_messages)
+
+        # Surface cross-session origin so downstream context observers can
+        # distinguish injected memory from content native to the current
+        # session. Loaded topic files may carry contributions from earlier
+        # sessions (tracked in ``MemoryTopicRecord.session_ids``); when any
+        # contributor differs from the current session, mark the injected
+        # block accordingly. See the ``CrossSessionObserver`` sample under
+        # ``samples/governance/cross_session_observer`` for an example
+        # subscriber, and the attribution mechanism on
+        # ``SessionContext.extend_messages``.
+        current_session_id = context.session_id
+        cross_session_origin: str | None = None
+        for record in selected_topics:
+            for contributor in record.session_ids:
+                if contributor and contributor != current_session_id:
+                    cross_session_origin = contributor
+                    break
+            if cross_session_origin is not None:
+                break
+
         context.extend_messages(
             self.source_id,
             [
@@ -1322,6 +1342,7 @@ async def consolidate_memories() -> str:
                     ],
                 )
             ],
+            origin_session_id=cross_session_origin,
         )
 
     async def after_run(
diff --git a/python/packages/core/agent_framework/_sessions.py b/python/packages/core/agent_framework/_sessions.py
index be4d4ea2856..322cd09f4ea 100644
--- a/python/packages/core/agent_framework/_sessions.py
+++ b/python/packages/core/agent_framework/_sessions.py
@@ -214,7 +214,13 @@ def response(self) -> AgentResponse | None:
         """The agent's response. Set by the framework after invocation, read-only for providers."""
         return self._response
 
-    def extend_messages(self, source: str | object, messages: Sequence[Message]) -> None:
+    def extend_messages(
+        self,
+        source: str | object,
+        messages: Sequence[Message],
+        *,
+        origin_session_id: str | None = None,
+    ) -> None:
         """Add context messages from a specific source.
 
         Messages are copied before attribution is added, so the caller's
@@ -229,6 +235,16 @@ def extend_messages(self, source: str | object, messages: Sequence[Message]) ->
                 object is passed, its class name is recorded as
                 ``source_type`` in the attribution.
             messages: The messages to add.
+            origin_session_id: Optional session_id that originally produced
+                these messages, when different from the current session. Set
+                by providers that inject content stored under a different
+                session than the requesting one (cross-session memory). The
+                value is exposed under ``additional_properties["_attribution"]
+                ["origin_session_id"]`` so downstream context observers can
+                detect cross-session content for governance, audit, or
+                behavioral-analysis purposes. Omit (default) when content
+                originates in the current session — absence of the field is
+                semantically equivalent to "no origin information."
         """
         if isinstance(source, str):
             source_id = source
@@ -236,6 +252,8 @@ def extend_messages(self, source: str | object, messages: Sequence[Message]) ->
         else:
             source_id = source.source_id  # type: ignore[attr-defined]
             attribution = {"source_id": source_id, "source_type": type(source).__name__}
+        if origin_session_id is not None:
+            attribution["origin_session_id"] = origin_session_id
 
         copied: list[Message] = []
         for message in messages:
diff --git a/python/packages/core/tests/core/test_harness_memory.py b/python/packages/core/tests/core/test_harness_memory.py
index 9d4c7c71d15..4b86467d9ff 100644
--- a/python/packages/core/tests/core/test_harness_memory.py
+++ b/python/packages/core/tests/core/test_harness_memory.py
@@ -500,6 +500,94 @@ async def test_memory_context_provider_recent_turns_can_skip_tool_call_groups(tm
     assert with_tools_messages[4].text == "Second final answer"
 
 
+async def test_memory_context_provider_marks_cross_session_origin(tmp_path) -> None:
+    """Injected memory should carry origin_session_id when topics originate in a prior session.
+
+    Exercises the cross-session attribution surface added to support downstream observers
+    detecting attacks of the class documented in Dai et al. (arXiv:2605.06158).
+    """
+    session = AgentSession(session_id="session-current")
+    session.state["owner_id"] = "alice"
+    store = MemoryFileStore(
+        tmp_path,
+        owner_state_key="owner_id",
+        dumps=lambda value: json.dumps(value, separators=(",", ":"), sort_keys=True),
+        loads=json.loads,
+    )
+    updated_at = datetime(2026, 4, 21, tzinfo=timezone.utc).replace(microsecond=0).isoformat()
+    store.write_topic(
+        session,
+        MemoryTopicRecord(
+            topic="travel preferences",
+            summary="Loves Oslo trips.",
+            memories=["Prefers Oslo in summer."],
+            updated_at=updated_at,
+            session_ids=["session-prior"],
+        ),
+        source_id=DEFAULT_MEMORY_SOURCE_ID,
+    )
+
+    agent = Agent(
+        client=_MemoryHarnessClient(),
+        context_providers=[MemoryContextProvider(store=store)],
+        default_options={"store": False},
+    )
+
+    session_context, _ = await agent._prepare_session_and_messages(  # type: ignore[reportPrivateUsage]
+        session=session,
+        input_messages=[Message(role="user", contents=["Tell me about my travel preferences."])],
+    )
+
+    memory_messages = [
+        m for m in session_context.context_messages.get(DEFAULT_MEMORY_SOURCE_ID, []) if "### MEMORY.md" in m.text
+    ]
+    assert memory_messages, "expected an injected memory block under the memory source"
+    attribution = memory_messages[0].additional_properties.get("_attribution") or {}
+    assert attribution.get("origin_session_id") == "session-prior"
+
+
+async def test_memory_context_provider_omits_origin_when_only_current_session(tmp_path) -> None:
+    """When all contributing topics are from the current session, attribution must NOT advertise an origin."""
+    session = AgentSession(session_id="session-current")
+    session.state["owner_id"] = "alice"
+    store = MemoryFileStore(
+        tmp_path,
+        owner_state_key="owner_id",
+        dumps=lambda value: json.dumps(value, separators=(",", ":"), sort_keys=True),
+        loads=json.loads,
+    )
+    updated_at = datetime(2026, 4, 21, tzinfo=timezone.utc).replace(microsecond=0).isoformat()
+    store.write_topic(
+        session,
+        MemoryTopicRecord(
+            topic="travel preferences",
+            summary="Loves Oslo trips.",
+            memories=["Prefers Oslo in summer."],
+            updated_at=updated_at,
+            session_ids=["session-current"],
+        ),
+        source_id=DEFAULT_MEMORY_SOURCE_ID,
+    )
+
+    agent = Agent(
+        client=_MemoryHarnessClient(),
+        context_providers=[MemoryContextProvider(store=store)],
+        default_options={"store": False},
+    )
+
+    session_context, _ = await agent._prepare_session_and_messages(  # type: ignore[reportPrivateUsage]
+        session=session,
+        input_messages=[Message(role="user", contents=["Tell me about my travel preferences."])],
+    )
+
+    memory_messages = [
+        m for m in session_context.context_messages.get(DEFAULT_MEMORY_SOURCE_ID, []) if "### MEMORY.md" in m.text
+    ]
+    assert memory_messages
+    attribution = memory_messages[0].additional_properties.get("_attribution") or {}
+    assert "origin_session_id" not in attribution
+
+
 async def test_memory_context_provider_uses_explicit_consolidation_client(tmp_path) -> None:
     """The memory provider should use the explicit consolidation client when one is configured."""
     session = AgentSession(session_id="session-1")
diff --git a/python/packages/core/tests/core/test_sessions.py b/python/packages/core/tests/core/test_sessions.py
index ebb91d0b0d0..808eec9aba3 100644
--- a/python/packages/core/tests/core/test_sessions.py
+++ b/python/packages/core/tests/core/test_sessions.py
@@ -107,6 +107,39 @@ class MyProvider:
         stored = ctx.context_messages["rag"][0]
         assert stored.additional_properties["_attribution"] == {"source_id": "rag", "source_type": "MyProvider"}
 
+    def test_extend_messages_origin_session_id_default_omits_field(self) -> None:
+        ctx = SessionContext(input_messages=[])
+        msg = Message(role="system", contents=["ctx"])
+        ctx.extend_messages("rag", [msg])
+        stored = ctx.context_messages["rag"][0]
+        # Default (no origin_session_id passed) preserves the historical attribution shape
+        # so observers can distinguish "no origin info" from "explicit cross-session marker."
+        assert "origin_session_id" not in stored.additional_properties["_attribution"]
+
+    def test_extend_messages_origin_session_id_recorded_on_attribution(self) -> None:
+        ctx = SessionContext(session_id="current", input_messages=[])
+        msg = Message(role="system", contents=["loaded from a prior session"])
+        ctx.extend_messages("memory_provider", [msg], origin_session_id="prior-session-id")
+        stored = ctx.context_messages["memory_provider"][0]
+        assert stored.additional_properties["_attribution"] == {
+            "source_id": "memory_provider",
+            "origin_session_id": "prior-session-id",
+        }
+
+    def test_extend_messages_origin_session_id_with_provider_object(self) -> None:
+        class MyMemoryProvider:
+            source_id = "memory"
+
+        ctx = SessionContext(session_id="current", input_messages=[])
+        msg = Message(role="assistant", contents=["consolidated memory content"])
+        ctx.extend_messages(MyMemoryProvider(), [msg], origin_session_id="prior")
+        stored = ctx.context_messages["memory"][0]
+        assert stored.additional_properties["_attribution"] == {
+            "source_id": "memory",
+            "source_type": "MyMemoryProvider",
+            "origin_session_id": "prior",
+        }
+
     def test_extend_instructions_string(self) -> None:
         ctx = SessionContext(input_messages=[])
         ctx.extend_instructions("sys", "Be helpful")
diff --git a/python/samples/02-agents/context_providers/README.md b/python/samples/02-agents/context_providers/README.md
index 04f3a1395f7..355cf4537d2 100644
--- a/python/samples/02-agents/context_providers/README.md
+++ b/python/samples/02-agents/context_providers/README.md
@@ -7,6 +7,7 @@ These samples demonstrate how to use context providers to enrich agent conversat
 | File / Folder | Description |
 |---------------|-------------|
 | [`simple_context_provider.py`](simple_context_provider.py) | Implement a custom context provider by extending `ContextProvider` to extract and inject structured user information across turns. |
+| [`cross_session_observer.py`](cross_session_observer.py) | Detect injected context messages whose origin differs from the current session, via the `_attribution["origin_session_id"]` field. Self-contained — no LLM credentials required. |
 | [`azure_ai_foundry_memory.py`](azure_ai_foundry_memory.py) | Use `FoundryMemoryProvider` to add semantic memory — automatically retrieves, searches, and stores memories via Azure AI Foundry. |
 | [`azure_ai_search/`](azure_ai_search/) | Retrieval Augmented Generation (RAG) with Azure AI Search in semantic and agentic modes. See its own [README](azure_ai_search/README.md). |
 | [`mem0/`](mem0/) | Memory-powered context using the Mem0 integration (open-source and managed). See its own [README](mem0/README.md). |
@@ -14,6 +15,9 @@ These samples demonstrate how to use context providers to enrich agent conversat
 
 ## Prerequisites
 
+**For `cross_session_observer.py`:**
+- No external dependencies; runs against in-memory `SessionContext`.
+
 **For `simple_context_provider.py`:**
 - `FOUNDRY_PROJECT_ENDPOINT`: Your Azure AI Foundry project endpoint
 - `FOUNDRY_MODEL`: Model deployment name
diff --git a/python/samples/02-agents/context_providers/cross_session_observer.py b/python/samples/02-agents/context_providers/cross_session_observer.py
new file mode 100644
index 00000000000..c502f9300e7
--- /dev/null
+++ b/python/samples/02-agents/context_providers/cross_session_observer.py
@@ -0,0 +1,141 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+from collections.abc import Mapping
+from typing import Any, Callable, cast
+
+from agent_framework import AgentSession, ContextProvider, Message, SessionContext
+
+"""This sample demonstrates how to detect cross-session memory injection.
+
+When a context provider injects messages from a different ``session_id`` than
+the requesting one — the legitimate cross-session memory use case (consolidated
+memories, Mem0 with default scope, shared knowledge bases) — the framework
+records the originating session under
+``message.additional_properties["_attribution"]["origin_session_id"]``.
+
+Downstream context observers can subscribe to this signal for governance,
+audit, and behavioral analysis purposes. This is useful for defending against
+the stateful-agent-backdoor attack class documented in Dai et al.,
+arXiv:2605.06158, in which an adversary chains sub-backdoors across sessions
+under permission isolation via persisted memory state.
+
+The sample is self-contained: it constructs ``SessionContext`` directly and
+invokes provider lifecycle methods manually, so no LLM credentials are
+required to run it.
+"""
+
+
+class CrossSessionObserver(ContextProvider):
+    """Detect injected context messages whose origin differs from the current session.
+
+    Subscribes via the standard ``ContextProvider`` pipeline. In ``before_run``,
+    walks the accumulated context messages and invokes a user-supplied
+    callback for each message whose ``_attribution["origin_session_id"]``
+    is present and differs from the current ``session_id``.
+
+    The callback receives the source_id that injected the content, the
+    originating session_id, the current session_id, and the message itself.
+    Use it to log, alert, increment metrics, or enforce policy — the observer
+    itself only surfaces the signal, leaving the response policy to the caller.
+    """
+
+    DEFAULT_SOURCE_ID = "cross_session_observer"
+
+    def __init__(
+        self,
+        on_cross_session_access: Callable[[str, str, str | None, Message], None],
+        *,
+        source_id: str = DEFAULT_SOURCE_ID,
+    ) -> None:
+        """Initialize the observer.
+
+        Args:
+            on_cross_session_access: Callback invoked for each detected
+                cross-session message. Signature is
+                ``(source_id, origin_session_id, current_session_id, message)``.
+            source_id: Unique identifier for this observer instance.
+        """
+        super().__init__(source_id)
+        self._on_cross_session_access = on_cross_session_access
+
+    async def before_run(
+        self,
+        *,
+        agent: Any,
+        session: AgentSession | None,
+        context: SessionContext,
+        state: dict[str, Any],
+    ) -> None:
+        """Inspect accumulated context messages for cross-session origin."""
+        current_session_id = context.session_id
+        for source_id, messages in context.context_messages.items():
+            if source_id == self.source_id:
+                continue
+            for message in messages:
+                attribution_raw = message.additional_properties.get("_attribution")
+                if not isinstance(attribution_raw, Mapping):
+                    continue
+                attribution = cast(Mapping[str, Any], attribution_raw)
+                origin = attribution.get("origin_session_id")
+                if isinstance(origin, str) and origin != current_session_id:
+                    self._on_cross_session_access(source_id, origin, current_session_id, message)
+
+
+def _on_detected(source_id: str, origin: str, current: str | None, message: Message) -> None:
+    """Sample callback that logs cross-session detections to stdout."""
+    preview = " ".join(message.text.split())[:80]
+    print(
+        f"[cross-session detected] source={source_id!r} "
+        f"origin_session={origin!r} current_session={current!r} "
+        f"preview={preview!r}"
+    )
+
+
+async def main() -> None:
+    """Demonstrate the observer firing on cross-session injection."""
+    observer = CrossSessionObserver(_on_detected)
+
+    # --- Case 1: same-session injection (observer should be silent) ---
+    same_session_context = SessionContext(
+        session_id="session-A",
+        input_messages=[Message("user", ["What did we discuss last time?"])],
+    )
+    # Simulate a same-session provider injecting same-session history. Omitting
+    # origin_session_id is semantically "no origin info" — observers treat as
+    # equivalent to same-session for backward compatibility.
+    same_session_context.extend_messages(
+        "history_provider",
+        [Message("assistant", ["We talked about Q3 revenue projections."])],
+    )
+    await observer.before_run(
+        agent=None,
+        session=None,
+        context=same_session_context,
+        state={},
+    )
+    print("--- Same-session case complete (no detections expected above) ---\n")
+
+    # --- Case 2: cross-session injection (observer should fire) ---
+    cross_session_context = SessionContext(
+        session_id="session-B",
+        input_messages=[Message("user", ["Continue from where we left off."])],
+    )
+    # Simulate a cross-session memory provider injecting content originally
+    # written in session-A while we're now running in session-B.
+    cross_session_context.extend_messages(
+        "memory_provider",
+        [Message("assistant", ["Remember: API key for prod is sk-7f3a... (from session-A)."])],
+        origin_session_id="session-A",
+    )
+    await observer.before_run(
+        agent=None,
+        session=None,
+        context=cross_session_context,
+        state={},
+    )
+    print("--- Cross-session case complete (one detection expected above) ---")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

From 6e924d5b1ee1f0f91e0828b09eacc873a5542a9b Mon Sep 17 00:00:00 2001
From: Evan Mattson <evan.mattson@microsoft.com>
Date: Fri, 10 Jul 2026 16:03:58 +0900
Subject: [PATCH 2/4] Address cross-session attribution review feedback

---
 python/packages/core/AGENTS.md                |  3 +-
 .../core/agent_framework/_harness/_memory.py  | 33 +++++++--------
 .../core/agent_framework/_sessions.py         | 38 ++++++++++-------
 .../core/tests/core/test_harness_memory.py    | 22 +++++-----
 .../packages/core/tests/core/test_sessions.py | 40 ++++++++++++++----
 .../02-agents/context_providers/README.md     |  2 +-
 .../cross_session_observer.py                 | 41 +++++++++++--------
 7 files changed, 107 insertions(+), 72 deletions(-)

diff --git a/python/packages/core/AGENTS.md b/python/packages/core/AGENTS.md
index 31124b70dff..83925f13878 100644
--- a/python/packages/core/AGENTS.md
+++ b/python/packages/core/AGENTS.md
@@ -75,7 +75,8 @@ agent_framework/
 
 - **`AgentSession`** - Manages conversation state and session metadata
 - **`ServiceSessionId`** - Mapping alias for structured service-owned continuation handles used in `AgentSession.service_session_id`
-- **`SessionContext`** - Context object for session-scoped data during agent runs
+- **`SessionContext`** - Context object for session-scoped data during agent runs. `extend_messages(...)` can attach
+  ordered, deduplicated `origin_session_ids` attribution when a provider injects content from other sessions.
 - **`ContextProvider`** - Base class for context providers (RAG, memory systems)
 - **`HistoryProvider`** - Base class for conversation history storage
 - **`InMemoryHistoryProvider`** - Built-in session-state history provider for local runs
diff --git a/python/packages/core/agent_framework/_harness/_memory.py b/python/packages/core/agent_framework/_harness/_memory.py
index a094299301b..dcfba6a9ef7 100644
--- a/python/packages/core/agent_framework/_harness/_memory.py
+++ b/python/packages/core/agent_framework/_harness/_memory.py
@@ -1307,24 +1307,21 @@ async def consolidate_memories() -> str:
         if recent_history_messages:
             context.extend_messages(self.source_id, recent_history_messages)
 
-        # Surface cross-session origin so downstream context observers can
-        # distinguish injected memory from content native to the current
-        # session. Loaded topic files may carry contributions from earlier
-        # sessions (tracked in ``MemoryTopicRecord.session_ids``); when any
-        # contributor differs from the current session, mark the injected
-        # block accordingly. See the ``CrossSessionObserver`` sample under
-        # ``samples/governance/cross_session_observer`` for an example
-        # subscriber, and the attribution mechanism on
-        # ``SessionContext.extend_messages``.
+        # Surface every cross-session origin so downstream context observers
+        # can distinguish injected memory from content native to the current
+        # session. Loaded topic files may carry contributions from multiple
+        # earlier sessions, tracked in ``MemoryTopicRecord.session_ids``.
+        # See ``samples/02-agents/context_providers/cross_session_observer.py``
+        # for an example subscriber.
         current_session_id = context.session_id
-        cross_session_origin: str | None = None
-        for record in selected_topics:
-            for contributor in record.session_ids:
-                if contributor and contributor != current_session_id:
-                    cross_session_origin = contributor
-                    break
-            if cross_session_origin is not None:
-                break
+        cross_session_origins = list(
+            dict.fromkeys(
+                contributor
+                for record in selected_topics
+                for contributor in record.session_ids
+                if contributor and contributor != current_session_id
+            )
+        )
 
         context.extend_messages(
             self.source_id,
@@ -1342,7 +1339,7 @@ async def consolidate_memories() -> str:
                     ],
                 )
             ],
-            origin_session_id=cross_session_origin,
+            origin_session_ids=cross_session_origins,
         )
 
     async def after_run(
diff --git a/python/packages/core/agent_framework/_sessions.py b/python/packages/core/agent_framework/_sessions.py
index eda1eb4367e..1e08c41716e 100644
--- a/python/packages/core/agent_framework/_sessions.py
+++ b/python/packages/core/agent_framework/_sessions.py
@@ -235,7 +235,7 @@ def extend_messages(
         source: str | object,
         messages: Sequence[Message],
         *,
-        origin_session_id: str | None = None,
+        origin_session_ids: Sequence[str] | None = None,
     ) -> None:
         """Add context messages from a specific source.
 
@@ -251,31 +251,41 @@ def extend_messages(
                 object is passed, its class name is recorded as
                 ``source_type`` in the attribution.
             messages: The messages to add.
-            origin_session_id: Optional session_id that originally produced
+            origin_session_ids: Optional session IDs that originally produced
                 these messages, when different from the current session. Set
-                by providers that inject content stored under a different
-                session than the requesting one (cross-session memory). The
-                value is exposed under ``additional_properties["_attribution"]
-                ["origin_session_id"]`` so downstream context observers can
-                detect cross-session content for governance, audit, or
-                behavioral-analysis purposes. Omit (default) when content
-                originates in the current session — absence of the field is
-                semantically equivalent to "no origin information."
+                by providers that inject content stored under other sessions
+                (cross-session memory). The values are exposed under
+                ``additional_properties["_attribution"]["origin_session_ids"]``
+                so downstream context observers can detect cross-session
+                content for governance, audit, or behavioral-analysis
+                purposes. Omit (default) when content originates in the
+                current session; absence of the field means that no origin
+                information was supplied.
         """
         if isinstance(source, str):
             source_id = source
-            attribution: dict[str, str] = {"source_id": source_id}
+            attribution: dict[str, Any] = {"source_id": source_id}
         else:
             source_id = source.source_id  # type: ignore[attr-defined]
             attribution = {"source_id": source_id, "source_type": type(source).__name__}
-        if origin_session_id is not None:
-            attribution["origin_session_id"] = origin_session_id
+        if origin_session_ids:
+            attribution["origin_session_ids"] = list(dict.fromkeys(origin_session_ids))
 
         copied: list[Message] = []
         for message in messages:
             msg_copy = copy.copy(message)
             msg_copy.additional_properties = dict(message.additional_properties)
-            msg_copy.additional_properties.setdefault("_attribution", attribution)
+            message_attribution = dict(attribution)
+            if "origin_session_ids" in message_attribution:
+                message_attribution["origin_session_ids"] = list(message_attribution["origin_session_ids"])
+            existing_attribution = msg_copy.additional_properties.get("_attribution")
+            if isinstance(existing_attribution, Mapping):
+                merged_attribution = dict(cast(Mapping[str, Any], existing_attribution))
+                for key, value in message_attribution.items():
+                    merged_attribution.setdefault(key, value)
+                msg_copy.additional_properties["_attribution"] = merged_attribution
+            else:
+                msg_copy.additional_properties.setdefault("_attribution", message_attribution)
             copied.append(msg_copy)
         if source_id not in self.context_messages:
             self.context_messages[source_id] = []
diff --git a/python/packages/core/tests/core/test_harness_memory.py b/python/packages/core/tests/core/test_harness_memory.py
index ecc6dd21683..f5b2733e577 100644
--- a/python/packages/core/tests/core/test_harness_memory.py
+++ b/python/packages/core/tests/core/test_harness_memory.py
@@ -505,8 +505,8 @@ async def test_memory_context_provider_recent_turns_can_skip_tool_call_groups(tm
     assert with_tools_messages[4].text == "Second final answer"
 
 
-async def test_memory_context_provider_marks_cross_session_origin(tmp_path) -> None:
-    """Injected memory should carry origin_session_id when topics originate in a prior session.
+async def test_memory_context_provider_marks_cross_session_origins(tmp_path) -> None:
+    """Injected memory should carry all prior session origins without duplicates.
 
     Exercises the cross-session attribution surface added to support downstream observers
     detecting attacks of the class documented in Dai et al. (arXiv:2605.06158).
@@ -527,15 +527,15 @@ async def test_memory_context_provider_marks_cross_session_origin(tmp_path) -> N
             summary="Loves Oslo trips.",
             memories=["Prefers Oslo in summer."],
             updated_at=updated_at,
-            session_ids=["session-prior"],
+            session_ids=["session-current", "session-prior-1", "session-prior-2", "session-prior-1"],
         ),
         source_id=DEFAULT_MEMORY_SOURCE_ID,
     )
 
     agent = Agent(
-        client=_MemoryHarnessClient(),
+        client=_MemoryHarnessClient(),  # type: ignore[arg-type]  # pyrefly: ignore[bad-argument-type]  # ty: ignore[invalid-argument-type]
         context_providers=[MemoryContextProvider(store=store)],
-        default_options={"store": False},
+        default_options=_no_store_options(),
     )
 
     session_context, _ = await agent._prepare_session_and_messages(  # type: ignore[reportPrivateUsage]
@@ -547,8 +547,8 @@ async def test_memory_context_provider_marks_cross_session_origin(tmp_path) -> N
         m for m in session_context.context_messages.get(DEFAULT_MEMORY_SOURCE_ID, []) if "### MEMORY.md" in m.text
     ]
     assert memory_messages, "expected an injected memory block under the memory source"
-    attribution = memory_messages[0].additional_properties.get("_attribution") or {}
-    assert attribution.get("origin_session_id") == "session-prior"
+    attribution: dict[str, Any] = memory_messages[0].additional_properties.get("_attribution") or {}
+    assert attribution.get("origin_session_ids") == ["session-prior-1", "session-prior-2"]
 
 
 async def test_memory_context_provider_omits_origin_when_only_current_session(tmp_path) -> None:
@@ -575,9 +575,9 @@ async def test_memory_context_provider_omits_origin_when_only_current_session(tm
     )
 
     agent = Agent(
-        client=_MemoryHarnessClient(),
+        client=_MemoryHarnessClient(),  # type: ignore[arg-type]  # pyrefly: ignore[bad-argument-type]  # ty: ignore[invalid-argument-type]
         context_providers=[MemoryContextProvider(store=store)],
-        default_options={"store": False},
+        default_options=_no_store_options(),
     )
 
     session_context, _ = await agent._prepare_session_and_messages(  # type: ignore[reportPrivateUsage]
@@ -589,8 +589,8 @@ async def test_memory_context_provider_omits_origin_when_only_current_session(tm
         m for m in session_context.context_messages.get(DEFAULT_MEMORY_SOURCE_ID, []) if "### MEMORY.md" in m.text
     ]
     assert memory_messages
-    attribution = memory_messages[0].additional_properties.get("_attribution") or {}
-    assert "origin_session_id" not in attribution
+    attribution: dict[str, Any] = memory_messages[0].additional_properties.get("_attribution") or {}
+    assert "origin_session_ids" not in attribution
 
 
 async def test_memory_context_provider_uses_explicit_consolidation_client(tmp_path) -> None:
diff --git a/python/packages/core/tests/core/test_sessions.py b/python/packages/core/tests/core/test_sessions.py
index e5fd923e41f..714a103c57e 100644
--- a/python/packages/core/tests/core/test_sessions.py
+++ b/python/packages/core/tests/core/test_sessions.py
@@ -107,38 +107,60 @@ class MyProvider:
         stored = ctx.context_messages["rag"][0]
         assert stored.additional_properties["_attribution"] == {"source_id": "rag", "source_type": "MyProvider"}
 
-    def test_extend_messages_origin_session_id_default_omits_field(self) -> None:
+    def test_extend_messages_origin_session_ids_default_omits_field(self) -> None:
         ctx = SessionContext(input_messages=[])
         msg = Message(role="system", contents=["ctx"])
         ctx.extend_messages("rag", [msg])
         stored = ctx.context_messages["rag"][0]
-        # Default (no origin_session_id passed) preserves the historical attribution shape
+        # Default (no origin_session_ids passed) preserves the historical attribution shape
         # so observers can distinguish "no origin info" from "explicit cross-session marker."
-        assert "origin_session_id" not in stored.additional_properties["_attribution"]
+        assert "origin_session_ids" not in stored.additional_properties["_attribution"]
 
-    def test_extend_messages_origin_session_id_recorded_on_attribution(self) -> None:
+    def test_extend_messages_origin_session_ids_recorded_on_attribution(self) -> None:
         ctx = SessionContext(session_id="current", input_messages=[])
         msg = Message(role="system", contents=["loaded from a prior session"])
-        ctx.extend_messages("memory_provider", [msg], origin_session_id="prior-session-id")
+        ctx.extend_messages(
+            "memory_provider",
+            [msg],
+            origin_session_ids=["prior-session-id", "another-session", "prior-session-id"],
+        )
         stored = ctx.context_messages["memory_provider"][0]
         assert stored.additional_properties["_attribution"] == {
             "source_id": "memory_provider",
-            "origin_session_id": "prior-session-id",
+            "origin_session_ids": ["prior-session-id", "another-session"],
         }
 
-    def test_extend_messages_origin_session_id_with_provider_object(self) -> None:
+    def test_extend_messages_origin_session_ids_with_provider_object(self) -> None:
         class MyMemoryProvider:
             source_id = "memory"
 
         ctx = SessionContext(session_id="current", input_messages=[])
         msg = Message(role="assistant", contents=["consolidated memory content"])
-        ctx.extend_messages(MyMemoryProvider(), [msg], origin_session_id="prior")
+        ctx.extend_messages(MyMemoryProvider(), [msg], origin_session_ids=["prior"])
         stored = ctx.context_messages["memory"][0]
         assert stored.additional_properties["_attribution"] == {
             "source_id": "memory",
             "source_type": "MyMemoryProvider",
-            "origin_session_id": "prior",
+            "origin_session_ids": ["prior"],
+        }
+
+    def test_extend_messages_adds_origin_to_existing_attribution(self) -> None:
+        ctx = SessionContext(session_id="current", input_messages=[])
+        msg = Message(
+            role="system",
+            contents=["loaded from a prior session"],
+            additional_properties={"_attribution": {"source_id": "custom", "custom_key": "value"}},
+        )
+
+        ctx.extend_messages("memory_provider", [msg], origin_session_ids=["prior"])
+
+        stored = ctx.context_messages["memory_provider"][0]
+        assert stored.additional_properties["_attribution"] == {
+            "source_id": "custom",
+            "custom_key": "value",
+            "origin_session_ids": ["prior"],
         }
+        assert msg.additional_properties["_attribution"] == {"source_id": "custom", "custom_key": "value"}
 
     def test_extend_instructions_string(self) -> None:
         ctx = SessionContext(input_messages=[])
diff --git a/python/samples/02-agents/context_providers/README.md b/python/samples/02-agents/context_providers/README.md
index 49417a6b06e..c82f3e82413 100644
--- a/python/samples/02-agents/context_providers/README.md
+++ b/python/samples/02-agents/context_providers/README.md
@@ -7,7 +7,7 @@ These samples demonstrate how to use context providers to enrich agent conversat
 | File / Folder | Description |
 |---------------|-------------|
 | [`simple_context_provider.py`](simple_context_provider.py) | Implement a custom context provider by extending `ContextProvider` to extract and inject structured user information across turns. |
-| [`cross_session_observer.py`](cross_session_observer.py) | Detect injected context messages whose origin differs from the current session, via the `_attribution["origin_session_id"]` field. Self-contained — no LLM credentials required. |
+| [`cross_session_observer.py`](cross_session_observer.py) | Detect injected context messages whose origins differ from the current session, via the `_attribution["origin_session_ids"]` field. Self-contained — no LLM credentials required. |
 | [`azure_ai_foundry_memory.py`](azure_ai_foundry_memory.py) | Use `FoundryMemoryProvider` to add semantic memory — automatically retrieves, searches, and stores memories via Azure AI Foundry. |
 | [`file_access_data_processing/`](file_access_data_processing/) | Use `FileAccessProvider` with `FileSystemAgentFileStore` to give an agent read/write/search access to a folder of CSV data files. See its own [README](file_access_data_processing/README.md). |
 | [`azure_ai_search/`](azure_ai_search/) | Retrieval Augmented Generation (RAG) with Azure AI Search in semantic and agentic modes. See its own [README](azure_ai_search/README.md). |
diff --git a/python/samples/02-agents/context_providers/cross_session_observer.py b/python/samples/02-agents/context_providers/cross_session_observer.py
index c502f9300e7..fe540da5fad 100644
--- a/python/samples/02-agents/context_providers/cross_session_observer.py
+++ b/python/samples/02-agents/context_providers/cross_session_observer.py
@@ -1,8 +1,8 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 import asyncio
-from collections.abc import Mapping
-from typing import Any, Callable, cast
+from collections.abc import Callable, Mapping, Sequence
+from typing import Any, cast
 
 from agent_framework import AgentSession, ContextProvider, Message, SessionContext
 
@@ -11,8 +11,8 @@
 When a context provider injects messages from a different ``session_id`` than
 the requesting one — the legitimate cross-session memory use case (consolidated
 memories, Mem0 with default scope, shared knowledge bases) — the framework
-records the originating session under
-``message.additional_properties["_attribution"]["origin_session_id"]``.
+records the originating sessions under
+``message.additional_properties["_attribution"]["origin_session_ids"]``.
 
 Downstream context observers can subscribe to this signal for governance,
 audit, and behavioral analysis purposes. This is useful for defending against
@@ -31,11 +31,11 @@ class CrossSessionObserver(ContextProvider):
 
     Subscribes via the standard ``ContextProvider`` pipeline. In ``before_run``,
     walks the accumulated context messages and invokes a user-supplied
-    callback for each message whose ``_attribution["origin_session_id"]``
-    is present and differs from the current ``session_id``.
+    callback for each message whose ``_attribution["origin_session_ids"]``
+    contains one or more sessions other than the current ``session_id``.
 
     The callback receives the source_id that injected the content, the
-    originating session_id, the current session_id, and the message itself.
+    originating session IDs, the current session_id, and the message itself.
     Use it to log, alert, increment metrics, or enforce policy — the observer
     itself only surfaces the signal, leaving the response policy to the caller.
     """
@@ -44,7 +44,7 @@ class CrossSessionObserver(ContextProvider):
 
     def __init__(
         self,
-        on_cross_session_access: Callable[[str, str, str | None, Message], None],
+        on_cross_session_access: Callable[[str, Sequence[str], str | None, Message], None],
         *,
         source_id: str = DEFAULT_SOURCE_ID,
     ) -> None:
@@ -53,7 +53,7 @@ def __init__(
         Args:
             on_cross_session_access: Callback invoked for each detected
                 cross-session message. Signature is
-                ``(source_id, origin_session_id, current_session_id, message)``.
+                ``(source_id, origin_session_ids, current_session_id, message)``.
             source_id: Unique identifier for this observer instance.
         """
         super().__init__(source_id)
@@ -77,17 +77,22 @@ async def before_run(
                 if not isinstance(attribution_raw, Mapping):
                     continue
                 attribution = cast(Mapping[str, Any], attribution_raw)
-                origin = attribution.get("origin_session_id")
-                if isinstance(origin, str) and origin != current_session_id:
-                    self._on_cross_session_access(source_id, origin, current_session_id, message)
+                origins = attribution.get("origin_session_ids")
+                if not isinstance(origins, Sequence) or isinstance(origins, str):
+                    continue
+                cross_session_origins = [
+                    origin for origin in origins if isinstance(origin, str) and origin != current_session_id
+                ]
+                if cross_session_origins:
+                    self._on_cross_session_access(source_id, cross_session_origins, current_session_id, message)
 
 
-def _on_detected(source_id: str, origin: str, current: str | None, message: Message) -> None:
+def _on_detected(source_id: str, origins: Sequence[str], current: str | None, message: Message) -> None:
     """Sample callback that logs cross-session detections to stdout."""
     preview = " ".join(message.text.split())[:80]
     print(
         f"[cross-session detected] source={source_id!r} "
-        f"origin_session={origin!r} current_session={current!r} "
+        f"origin_sessions={list(origins)!r} current_session={current!r} "
         f"preview={preview!r}"
     )
 
@@ -102,7 +107,7 @@ async def main() -> None:
         input_messages=[Message("user", ["What did we discuss last time?"])],
     )
     # Simulate a same-session provider injecting same-session history. Omitting
-    # origin_session_id is semantically "no origin info" — observers treat as
+    # Omitting origin_session_ids means "no origin info"; observers treat it as
     # equivalent to same-session for backward compatibility.
     same_session_context.extend_messages(
         "history_provider",
@@ -122,11 +127,11 @@ async def main() -> None:
         input_messages=[Message("user", ["Continue from where we left off."])],
     )
     # Simulate a cross-session memory provider injecting content originally
-    # written in session-A while we're now running in session-B.
+    # written in sessions A and C while we're now running in session B.
     cross_session_context.extend_messages(
         "memory_provider",
-        [Message("assistant", ["Remember: API key for prod is sk-7f3a... (from session-A)."])],
-        origin_session_id="session-A",
+        [Message("assistant", ["Remember: API key for prod is sk-7f3a... (from prior sessions)."])],
+        origin_session_ids=["session-A", "session-C"],
     )
     await observer.before_run(
         agent=None,

From ea36b94d6d244bfddf2136af2dd24249c6dce4bb Mon Sep 17 00:00:00 2001
From: Evan Mattson <evan.mattson@microsoft.com>
Date: Fri, 10 Jul 2026 16:34:54 +0900
Subject: [PATCH 3/4] Address follow-up review feedback

---
 .../packages/core/agent_framework/_sessions.py | 14 +++++++++++++-
 .../packages/core/tests/core/test_sessions.py  | 18 ++++++++++++++----
 .../02-agents/context_providers/README.md      |  2 +-
 .../cross_session_observer.py                  |  6 +++---
 4 files changed, 31 insertions(+), 9 deletions(-)

diff --git a/python/packages/core/agent_framework/_sessions.py b/python/packages/core/agent_framework/_sessions.py
index 1e08c41716e..bf7ca140a68 100644
--- a/python/packages/core/agent_framework/_sessions.py
+++ b/python/packages/core/agent_framework/_sessions.py
@@ -282,7 +282,19 @@ def extend_messages(
             if isinstance(existing_attribution, Mapping):
                 merged_attribution = dict(cast(Mapping[str, Any], existing_attribution))
                 for key, value in message_attribution.items():
-                    merged_attribution.setdefault(key, value)
+                    if key == "origin_session_ids":
+                        existing_origins = merged_attribution.get(key)
+                        if isinstance(existing_origins, Sequence) and not isinstance(existing_origins, str):
+                            existing_origin_values = cast(Sequence[Any], existing_origins)
+                            value = list(
+                                dict.fromkeys(
+                                    [origin for origin in existing_origin_values if isinstance(origin, str)]
+                                    + cast(list[str], value)
+                                )
+                            )
+                        merged_attribution[key] = value
+                    else:
+                        merged_attribution.setdefault(key, value)
                 msg_copy.additional_properties["_attribution"] = merged_attribution
             else:
                 msg_copy.additional_properties.setdefault("_attribution", message_attribution)
diff --git a/python/packages/core/tests/core/test_sessions.py b/python/packages/core/tests/core/test_sessions.py
index 714a103c57e..69f7fca8056 100644
--- a/python/packages/core/tests/core/test_sessions.py
+++ b/python/packages/core/tests/core/test_sessions.py
@@ -149,18 +149,28 @@ def test_extend_messages_adds_origin_to_existing_attribution(self) -> None:
         msg = Message(
             role="system",
             contents=["loaded from a prior session"],
-            additional_properties={"_attribution": {"source_id": "custom", "custom_key": "value"}},
+            additional_properties={
+                "_attribution": {
+                    "source_id": "custom",
+                    "custom_key": "value",
+                    "origin_session_ids": ["existing", "prior"],
+                }
+            },
         )
 
-        ctx.extend_messages("memory_provider", [msg], origin_session_ids=["prior"])
+        ctx.extend_messages("memory_provider", [msg], origin_session_ids=["prior", "new"])
 
         stored = ctx.context_messages["memory_provider"][0]
         assert stored.additional_properties["_attribution"] == {
             "source_id": "custom",
             "custom_key": "value",
-            "origin_session_ids": ["prior"],
+            "origin_session_ids": ["existing", "prior", "new"],
+        }
+        assert msg.additional_properties["_attribution"] == {
+            "source_id": "custom",
+            "custom_key": "value",
+            "origin_session_ids": ["existing", "prior"],
         }
-        assert msg.additional_properties["_attribution"] == {"source_id": "custom", "custom_key": "value"}
 
     def test_extend_instructions_string(self) -> None:
         ctx = SessionContext(input_messages=[])
diff --git a/python/samples/02-agents/context_providers/README.md b/python/samples/02-agents/context_providers/README.md
index c82f3e82413..f860230f8cd 100644
--- a/python/samples/02-agents/context_providers/README.md
+++ b/python/samples/02-agents/context_providers/README.md
@@ -7,7 +7,7 @@ These samples demonstrate how to use context providers to enrich agent conversat
 | File / Folder | Description |
 |---------------|-------------|
 | [`simple_context_provider.py`](simple_context_provider.py) | Implement a custom context provider by extending `ContextProvider` to extract and inject structured user information across turns. |
-| [`cross_session_observer.py`](cross_session_observer.py) | Detect injected context messages whose origins differ from the current session, via the `_attribution["origin_session_ids"]` field. Self-contained — no LLM credentials required. |
+| [`cross_session_observer.py`](cross_session_observer.py) | Detect injected context messages whose origins differ from the current session, via the `Message.additional_properties["_attribution"]["origin_session_ids"]` field. Self-contained — no LLM credentials required. |
 | [`azure_ai_foundry_memory.py`](azure_ai_foundry_memory.py) | Use `FoundryMemoryProvider` to add semantic memory — automatically retrieves, searches, and stores memories via Azure AI Foundry. |
 | [`file_access_data_processing/`](file_access_data_processing/) | Use `FileAccessProvider` with `FileSystemAgentFileStore` to give an agent read/write/search access to a folder of CSV data files. See its own [README](file_access_data_processing/README.md). |
 | [`azure_ai_search/`](azure_ai_search/) | Retrieval Augmented Generation (RAG) with Azure AI Search in semantic and agentic modes. See its own [README](azure_ai_search/README.md). |
diff --git a/python/samples/02-agents/context_providers/cross_session_observer.py b/python/samples/02-agents/context_providers/cross_session_observer.py
index fe540da5fad..67554b3351d 100644
--- a/python/samples/02-agents/context_providers/cross_session_observer.py
+++ b/python/samples/02-agents/context_providers/cross_session_observer.py
@@ -107,8 +107,8 @@ async def main() -> None:
         input_messages=[Message("user", ["What did we discuss last time?"])],
     )
     # Simulate a same-session provider injecting same-session history. Omitting
-    # Omitting origin_session_ids means "no origin info"; observers treat it as
-    # equivalent to same-session for backward compatibility.
+    # origin_session_ids means "no origin info"; observers treat it as equivalent
+    # to same-session for backward compatibility.
     same_session_context.extend_messages(
         "history_provider",
         [Message("assistant", ["We talked about Q3 revenue projections."])],
@@ -130,7 +130,7 @@ async def main() -> None:
     # written in sessions A and C while we're now running in session B.
     cross_session_context.extend_messages(
         "memory_provider",
-        [Message("assistant", ["Remember: API key for prod is sk-7f3a... (from prior sessions)."])],
+        [Message("assistant", ["Remember: API key for prod is <REDACTED> (from prior sessions)."])],
         origin_session_ids=["session-A", "session-C"],
     )
     await observer.before_run(

From f4f922c48a7780152ac20b3bb88516a545e56fe1 Mon Sep 17 00:00:00 2001
From: Evan Mattson <evan.mattson@microsoft.com>
Date: Sat, 11 Jul 2026 07:49:04 +0900
Subject: [PATCH 4/4] Address follow-up review comments

---
 python/packages/core/agent_framework/_sessions.py      |  2 ++
 .../context_providers/cross_session_observer.py        | 10 ++++++++++
 2 files changed, 12 insertions(+)

diff --git a/python/packages/core/agent_framework/_sessions.py b/python/packages/core/agent_framework/_sessions.py
index bf7ca140a68..5969704af83 100644
--- a/python/packages/core/agent_framework/_sessions.py
+++ b/python/packages/core/agent_framework/_sessions.py
@@ -251,6 +251,8 @@ def extend_messages(
                 object is passed, its class name is recorded as
                 ``source_type`` in the attribution.
             messages: The messages to add.
+
+        Keyword Args:
             origin_session_ids: Optional session IDs that originally produced
                 these messages, when different from the current session. Set
                 by providers that inject content stored under other sessions
diff --git a/python/samples/02-agents/context_providers/cross_session_observer.py b/python/samples/02-agents/context_providers/cross_session_observer.py
index 67554b3351d..dac88f39382 100644
--- a/python/samples/02-agents/context_providers/cross_session_observer.py
+++ b/python/samples/02-agents/context_providers/cross_session_observer.py
@@ -144,3 +144,13 @@ async def main() -> None:
 
 if __name__ == "__main__":
     asyncio.run(main())
+
+
+"""
+Sample output:
+--- Same-session case complete (no detections expected above) ---
+
+[cross-session detected] source='memory_provider' origin_sessions=['session-A', 'session-C'] \
+current_session='session-B' preview='Remember: API key for prod is <REDACTED> (from prior sessions).'
+--- Cross-session case complete (one detection expected above) ---
+"""