Python: feat: cross-session origin attribution on context messages

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

python/packages/core/agent_framework/_harness/_memory.py

@@ -1306,6 +1306,23 @@ async def consolidate_memories() -> str:
         )
         if recent_history_messages:
             context.extend_messages(self.source_id, recent_history_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_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,
             [
@@ -1322,6 +1339,7 @@ async def consolidate_memories() -> str:
                     ],
                 )
             ],
+            origin_session_ids=cross_session_origins,
         )
 
     async def after_run(

python/packages/core/agent_framework/_sessions.py

@@ -230,7 +230,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_ids: Sequence[str] | None = None,
+    ) -> None:
         """Add context messages from a specific source.
 
         Messages are copied before attribution is added, so the caller's
@@ -245,19 +251,55 @@ 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.
+
+        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
+                (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_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():
+                    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)
             copied.append(msg_copy)
         if source_id not in self.context_messages:
             self.context_messages[source_id] = []

python/packages/core/tests/core/test_harness_memory.py

@@ -505,6 +505,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_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).
+    """
+    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", "session-prior-1", "session-prior-2", "session-prior-1"],
+        ),
+        source_id=DEFAULT_MEMORY_SOURCE_ID,
+    )
+
+    agent = Agent(
+        client=_MemoryHarnessClient(),  # type: ignore[arg-type]  # pyrefly: ignore[bad-argument-type]  # ty: ignore[invalid-argument-type]
+        context_providers=[MemoryContextProvider(store=store)],
+        default_options=_no_store_options(),
+    )
+
+    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: 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:
+    """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(),  # type: ignore[arg-type]  # pyrefly: ignore[bad-argument-type]  # ty: ignore[invalid-argument-type]
+        context_providers=[MemoryContextProvider(store=store)],
+        default_options=_no_store_options(),
+    )
+
+    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: 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:
     """The memory provider should use the explicit consolidation client when one is configured."""
     session = AgentSession(session_id="session-1")

python/packages/core/tests/core/test_sessions.py

@@ -107,6 +107,71 @@ 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_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_ids passed) preserves the historical attribution shape
+        # so observers can distinguish "no origin info" from "explicit cross-session marker."
+        assert "origin_session_ids" not in stored.additional_properties["_attribution"]
+
+    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_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_ids": ["prior-session-id", "another-session"],
+        }
+
+    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_ids=["prior"])
+        stored = ctx.context_messages["memory"][0]
+        assert stored.additional_properties["_attribution"] == {
+            "source_id": "memory",
+            "source_type": "MyMemoryProvider",
+            "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",
+                    "origin_session_ids": ["existing", "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": ["existing", "prior", "new"],
+        }
+        assert msg.additional_properties["_attribution"] == {
+            "source_id": "custom",
+            "custom_key": "value",
+            "origin_session_ids": ["existing", "prior"],
+        }
+
     def test_extend_instructions_string(self) -> None:
         ctx = SessionContext(input_messages=[])
         ctx.extend_instructions("sys", "Be helpful")

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 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). |
@@ -15,6 +16,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