@@ -442,7 +442,8 @@ async def _run_compaction_locked(
         if resolved_mode == "previous_response_id":
             compact_kwargs["previous_response_id"] = self._response_id
         else:
-            compact_kwargs["input"] = session_items
+            # Manual compaction may reuse cached raw output from the preceding compaction.
+            compact_kwargs["input"] = normalize_input_items_for_api(session_items)
 
         compacted = await self.client.responses.compact(**compact_kwargs)
 
 
@@ -713,13 +713,15 @@ def _dedupe_key(item: TResponseInputItem) -> str | None:
 
 
 def strip_internal_input_item_metadata(item: TResponseInputItem) -> TResponseInputItem:
-    """Remove SDK-only session metadata before sending items back to the model."""
+    """Remove session and output-only metadata before replaying items as input."""
     if not isinstance(item, dict):
         return item
 
     cleaned = dict(item)
     cleaned.pop(TOOL_CALL_SESSION_DESCRIPTION_KEY, None)
     cleaned.pop(TOOL_CALL_SESSION_TITLE_KEY, None)
+    # Compaction and persisted history can bypass RunItem's output-to-input conversion.
+    cleaned.pop("created_by", None)
     return cast(TResponseInputItem, cleaned)
 
 
 
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import logging
 import sqlite3
 import threading
@@ -11,15 +12,17 @@
 from typing import Any, cast
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import httpx2
 import pytest
+from openai import AsyncOpenAI
 from openai.types.responses.response_usage import (
     InputTokensDetails,
     OutputTokensDetails,
     ResponseUsage,
 )
 
 import agents._debug as _debug
-from agents import Agent, Runner
+from agents import Agent, OpenAIResponsesModel, RunConfig, Runner
 from agents.items import MessageOutputItem, TResponseInputItem
 from agents.memory import (
     OpenAIResponsesCompactionSession,
@@ -45,10 +48,94 @@
 )
 from agents.run_state import RunState
 from agents.testing import ModelStep, ScriptedModel
+from tests.model_test_helpers import get_response_obj
 from tests.test_responses import get_function_tool, get_function_tool_call, get_text_message
 from tests.utils.simple_session import SimpleListSession
 
 
+@pytest.mark.asyncio
+@pytest.mark.allow_call_model_methods
+@pytest.mark.parametrize("history_source", ["automatic", "manual", "stored"])
+async def test_compaction_replay_strips_created_by(history_source: str) -> None:
+    compacted_item = {
+        "id": "cmp_synthetic",
+        "type": "compaction",
+        "encrypted_content": "synthetic-encrypted-content",
+        "created_by": "server",
+    }
+    request_inputs: list[list[dict[str, Any]]] = []
+
+    async def handler(request: httpx2.Request) -> httpx2.Response:
+        request_inputs.append(json.loads(request.content)["input"])
+        if request.url.path.endswith("/compact"):
+            return httpx2.Response(
+                200,
+                json={
+                    "id": "cmp_response",
+                    "object": "response.compaction",
+                    "created_at": 1,
+                    "output": [compacted_item],
+                },
+            )
+        assert request.url.path.endswith("/responses")
+        return httpx2.Response(
+            200, content=get_response_obj([get_text_message("42")]).model_dump_json()
+        )
+
+    underlying = SQLiteSession("created-by-replay")
+    try:
+        async with AsyncOpenAI(
+            api_key="synthetic-key",
+            http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(handler)),
+        ) as client:
+            session = OpenAIResponsesCompactionSession(
+                session_id="created-by-replay",
+                underlying_session=underlying,
+                client=client,
+                model="gpt-5",
+                compaction_mode="input",
+                should_trigger_compaction=lambda ctx: bool(ctx["compaction_candidate_items"]),
+            )
+            agent = Agent(name="test", model=OpenAIResponsesModel("gpt-5", openai_client=client))
+            config = RunConfig(tracing_disabled=True)
+            if history_source == "stored":
+                # History saved before the fix must also remain usable.
+                await underlying.add_items([cast(TResponseInputItem, compacted_item)])
+            elif history_source == "manual":
+                await session.add_items([{"role": "user", "content": "How many?"}])
+                await session.run_compaction({"force": True})
+            else:
+                await Runner.run(agent, "How many?", session=session, run_config=config)
+
+            stored_items = await underlying.get_items()
+            assert stored_items == [compacted_item]
+            if history_source == "manual":
+                # Repeated manual compaction reuses cached output without a history reload.
+                await session.add_items([{"role": "user", "content": "More context"}])
+                await session.run_compaction({"force": True})
+            result = await Runner.run(agent, "And urgent?", session=session, run_config=config)
+            assert result.final_output == "42"
+
+            # Inspect encoded inputs to both Responses and the next compact request.
+            replayed_items = [
+                item
+                for items in request_inputs
+                for item in items
+                if item.get("type") == "compaction"
+            ]
+            assert replayed_items == [
+                {
+                    "id": "cmp_synthetic",
+                    "type": "compaction",
+                    "encrypted_content": "synthetic-encrypted-content",
+                }
+            ] * (3 if history_source == "manual" else 2)
+            assert stored_items[0] == compacted_item
+            assert (await underlying.get_items())[0] == compacted_item
+    finally:
+        underlying.close()
+
+
 class TestIsOpenAIModelName:
     def test_gpt_models(self) -> None:
         assert is_openai_model_name("gpt-4o") is True