diff --git a/python/beeai_framework/tools/scratchpad/__init__.py b/python/beeai_framework/tools/scratchpad/__init__.py
new file mode 100644
index 000000000..b7e70bf0a
--- /dev/null
+++ b/python/beeai_framework/tools/scratchpad/__init__.py
@@ -0,0 +1,6 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from beeai_framework.tools.scratchpad.scratchpad import ScratchpadInput, ScratchpadTool
+
+__all__ = ["ScratchpadInput", "ScratchpadTool"]
diff --git a/python/beeai_framework/tools/scratchpad/scratchpad.py b/python/beeai_framework/tools/scratchpad/scratchpad.py
new file mode 100644
index 000000000..4da10d450
--- /dev/null
+++ b/python/beeai_framework/tools/scratchpad/scratchpad.py
@@ -0,0 +1,432 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+"""
+Agent Scratchpad Tool - Allows agents to track their reasoning and actions.
+
+This tool provides a working memory (scratchpad) where agents can:
+- Record actions they've taken
+- Store observations/results from tools
+- Review their previous reasoning
+- Avoid repeating actions
+"""
+
+import asyncio
+from beeai_framework.logger import Logger
+import re
+from typing import ClassVar
+
+from pydantic import BaseModel, Field
+
+from beeai_framework.context import RunContext
+from beeai_framework.emitter import Emitter
+from beeai_framework.tools import (
+    StringToolOutput,
+    Tool,
+    ToolInputValidationError,
+    ToolRunOptions,
+)
+
+logger = Logger(__name__)
+
+
+class ScratchpadInput(BaseModel):
+    """Input schema for scratchpad operations."""
+
+    operation: str = Field(
+        description=(
+            "Operation to perform: 'read' to view scratchpad, "
+            "'write' to add entry, 'append' to add to last entry, "
+            "'clear' to reset"
+        ),
+        enum=["read", "write", "append", "clear"],
+    )
+    content: str | None = Field(
+        default=None,
+        description=(
+            "Content to write/append (required for 'write' and 'append' " "operations)"
+        ),
+    )
+
+
+class ScratchpadTool(Tool):
+    """Tool for managing agent scratchpad (working memory).
+
+    Supports two types of content:
+    1. Key-Value Pairs: Automatically consolidated into a single entry
+       - Format: "key: value, another_key: another_value"
+       - Duplicate keys are updated with latest values
+       - Results in ONE entry containing all key-value pairs
+
+    2. Plain Text: Appended as separate entries
+       - Format: Any text without colons
+       - Each write creates a new list entry
+
+    This design allows structured state management (key-value) while
+    preserving free-form notes (plain text) as separate items.
+
+    Design Note (Lifecycle Management):
+    This tool uses a class-level dictionary (`_scratchpads`) for storage. In a
+    long-running process, this dictionary can grow indefinitely. Consumers
+    should ensure that `clear_session(session_id)` is called when a session
+    or agent run is complete to prevent memory leaks. For distributed
+    deployments, consider a persistent external store instead.
+    """
+
+    _scratchpads: ClassVar[dict[str, list[str]]] = {}
+    _lock: ClassVar[asyncio.Lock] = asyncio.Lock()
+
+    def __init__(self) -> None:
+        """Initialize scratchpad tool."""
+        super().__init__()
+        self.middlewares = []
+
+    @classmethod
+    def _ensure_session(cls, session_id: str) -> None:
+        """Ensure a session exists in scratchpads."""
+        if session_id not in cls._scratchpads:
+            cls._scratchpads[session_id] = []
+
+    def _get_session_id(self) -> str:
+        """Extract a stable session ID from RunContext.
+
+        Order of preference:
+        1) RunContext.context["session_id"] (set by the API layer)
+        2) RunContext.group_id (stable for a run group)
+        3) RunContext.run_id (per-request fallback)
+        """
+        context = RunContext.get()
+        if not context:
+            raise ToolInputValidationError("RunContext missing; cannot determine session.")
+
+        context_data = getattr(context, "context", None)
+        if isinstance(context_data, dict):
+            session_id = context_data.get("session_id")
+            if session_id:
+                return str(session_id)
+
+        group_id = getattr(context, "group_id", None)
+        if group_id:
+            return str(group_id)
+
+        run_id = getattr(context, "run_id", None)
+        if run_id:
+            return str(run_id)
+
+        raise ToolInputValidationError("No valid session id found in RunContext.")
+
+    @property
+    def name(self) -> str:
+        """Tool name."""
+        return "scratchpad"
+
+    @property
+    def description(self) -> str:
+        """Tool description."""
+        return (
+            "Manage your working memory (scratchpad). Use this to track "
+            "what you've done, what results you got, and avoid repeating "
+            "actions. Operations: 'read' - see your scratchpad, 'write' - "
+            "add an entry, 'clear' - reset scratchpad, 'append' - add to "
+            "existing entry."
+        )
+
+    @property
+    def input_schema(self) -> type[ScratchpadInput]:
+        """Input schema for the tool."""
+        return ScratchpadInput
+
+    def _create_emitter(self) -> Emitter:
+        """Create emitter for the tool."""
+        return Emitter.root()
+
+    @property
+    def emitter(self) -> Emitter:
+        """Emitter for the tool."""
+        return Emitter.root().child(
+            namespace=["tool", "scratchpad"],
+            creator=self,
+        )
+
+    def _get_entries(self, session_id: str) -> list[str]:
+        """Get scratchpad entries for a session.
+
+        Args:
+            session_id: Session identifier.
+
+        Returns:
+            List of scratchpad entries.
+        """
+        self._ensure_session(session_id)
+        return self._scratchpads[session_id]
+
+    def _read_scratchpad(self, session_id: str) -> str:
+        """Read the current scratchpad content.
+
+        Args:
+            session_id: Session identifier.
+
+        Returns:
+            Formatted scratchpad content string.
+        """
+        entries = self._get_entries(session_id)
+        if not entries:
+            result = "Scratchpad is empty. No actions recorded yet."
+            logger.info(f"ScratchpadTool[{session_id}]: READ - Empty")
+            return result
+
+        result = "=== AGENT SCRATCHPAD ===\n\n"
+        result += "\n\n".join(f"[{i}] {entry}" for i, entry in enumerate(entries, 1))
+
+        logger.info(f"ScratchpadTool[{session_id}]: READ - {len(entries)} entries")
+        return result
+
+    @staticmethod
+    def _parse_key_value_pairs(content: str) -> dict:
+        """Parse key-value pairs from scratchpad content.
+
+        Uses regex to correctly handle values containing commas, which prevents
+        incorrectly splitting "item: milk, bread, eggs" into separate entries.
+
+        Format Examples:
+        - Simple: "key: value" → {"key": "value"}
+        - Multiple: "key1: val1, key2: val2" → {"key1": "val1", "key2": "val2"}
+        - Comma in value: "item: milk, bread, key2: val2" → {"item": "milk, bread", "key2": "val2"}
+        - Hyphenated keys: "Content-Type: json, user-id: 123" → {"Content-Type": "json", "user-id": "123"}
+
+        Implementation Note:
+        A simple split-by-comma approach fails when values contain commas.
+        The regex pattern works by:
+        1. Matching any characters except ':' as the key: ([^:]+)
+        2. Matching the colon separator: :
+        3. Capturing everything until the next "key:" pattern or end: (.*?)(?=...)
+
+        This ensures commas within values are preserved while correctly
+        identifying multiple key-value pairs separated by commas.
+
+        Args:
+            content: Content string to parse.
+
+        Returns:
+            Dictionary of key-value pairs.
+        """
+        pairs = {}
+
+        # Regex breakdown:
+        # ([^:]+)           - Capture key (any chars except colon)
+        # :\s*              - Match colon and optional whitespace
+        # (.*?)             - Capture value (non-greedy)
+        # (?=               - Lookahead (doesn't consume characters):
+        #   \s*,\s*[^:]+:   - Next key-value pair (comma, then key:)
+        #   |               - OR
+        #   \s*$            - End of string
+        # )
+        pattern = re.compile(r"([^:]+):\s*(.*?)(?=\s*,\s*[^:]+:|\s*$)")
+
+        for match in pattern.finditer(content):
+            key = match.group(1).strip()
+            value = match.group(2).strip()
+            if key and value:
+                pairs[key] = value
+
+        return pairs
+
+    @staticmethod
+    def _merge_entries(entries: list[str], new_pairs: dict) -> list[str]:
+        """Merge new key-value pairs into existing entries.
+
+        IMPORTANT BEHAVIOR:
+        This method consolidates ALL key-value pairs (existing + new) into a
+        SINGLE entry. This means the returned list will contain at most ONE
+        consolidated entry, not multiple separate entries.
+
+        Design Rationale:
+        - Key-value pairs represent structured state that should be merged
+        - Each key appears only once with its latest value
+        - Prevents duplicate keys and maintains a single source of truth
+        - Example: Writing "city: Boston" then "date: 2025-01-28" results in
+          ONE entry: "city: Boston, date: 2025-01-28"
+
+        This is different from non-key-value entries (plain text) which are
+        appended as separate list items.
+
+        Args:
+            entries: List of existing scratchpad entries.
+            new_pairs: Dictionary of new key-value pairs to merge.
+
+        Returns:
+            Updated list with a SINGLE consolidated entry containing all
+            key-value pairs (old + new), with new values overriding old
+            values for duplicate keys. Returns empty list if no valid pairs.
+        """
+        # Parse all existing entries into a single dict
+        consolidated = {}
+        for entry in entries:
+            pairs = ScratchpadTool._parse_key_value_pairs(entry)
+            consolidated.update(pairs)
+
+        # Merge new pairs (new values override old ones for duplicate keys)
+        consolidated.update(new_pairs)
+
+        # Convert back to entry format
+        if consolidated:
+            # Create a single consolidated entry containing all pairs
+            entry_str = ", ".join(f"{k}: {v}" for k, v in consolidated.items())
+            return [entry_str]
+        return []
+
+    def _write_scratchpad(self, entry: str, session_id: str) -> str:
+        """Add or update entry in the scratchpad.
+
+        Behavior depends on entry format:
+
+        1. Key-Value Pairs (contains ":"):
+           - Parsed and merged with existing key-value pairs
+           - Results in a SINGLE consolidated entry
+           - New values override old values for duplicate keys
+
+        2. Plain Text (no ":"):
+           - Appended as a new separate entry
+           - Multiple plain text entries can exist
+
+        This design ensures structured state (key-value) remains consolidated
+        while allowing free-form notes (plain text) to accumulate.
+
+        Args:
+            entry: Content to add/update.
+            session_id: Session identifier.
+
+        Returns:
+            Success message describing the action taken.
+        """
+        entries = self._get_entries(session_id)
+        new_pairs = self._parse_key_value_pairs(entry)
+
+        if new_pairs:
+            # Merge with existing entries
+            entries[:] = self._merge_entries(entries, new_pairs)
+            # Since we just merged new_pairs (which is not empty), entries will have content
+            result = f"Updated scratchpad to: {entries[0]}"
+        else:
+            # If no key-value pairs found, append as-is (for non-structured entries)
+            entries.append(entry)
+            result = f"Added to scratchpad: {entry}"
+
+        logger.info(
+            f"ScratchpadTool[{session_id}]: WRITE - " f"{len(entries)} total entries"
+        )
+        return result
+
+    def _append_scratchpad(self, text: str, session_id: str) -> str:
+        """Append to the last entry in scratchpad.
+
+        Args:
+            text: Text to append.
+            session_id: Session identifier.
+
+        Returns:
+            Success or error message.
+        """
+        entries = self._get_entries(session_id)
+        if not entries:
+            result = "No entry to append to. Use 'write' first."
+            logger.info(f"ScratchpadTool[{session_id}]: APPEND - No entries")
+            return result
+
+        entries[-1] += f" {text}"
+        result = f"Appended to last entry: {text}"
+        logger.info(f"ScratchpadTool[{session_id}]: APPEND - Updated")
+        return result
+
+    def _clear_scratchpad(self, session_id: str) -> str:
+        """Clear the scratchpad.
+
+        Args:
+            session_id: Session identifier.
+
+        Returns:
+            Success message.
+        """
+        entries_count = len(self._get_entries(session_id))
+        self._scratchpads[session_id] = []
+        result = "Scratchpad cleared."
+        logger.info(
+            f"ScratchpadTool[{session_id}]: CLEAR - " f"{entries_count} entries"
+        )
+        return result
+
+    async def _run(
+        self,
+        input: ScratchpadInput,
+        options: ToolRunOptions | None = None,
+    ) -> StringToolOutput:
+        """Execute scratchpad operation.
+
+        Args:
+            input: ScratchpadInput model instance.
+            options: Optional tool run options.
+
+        Returns:
+            StringToolOutput with the result of the operation.
+        """
+        session_id = self._get_session_id()
+        operation = input.operation.lower().strip()
+        content = input.content
+
+        logger.info(
+            f"ScratchpadTool[{session_id}]: operation='{operation}', "
+            f"content='{content}'"
+        )
+
+        result = None
+        async with ScratchpadTool._lock:
+            if operation in ("write", "append") and not content:
+                self._raise_input_validation_error(
+                    f"'{operation}' operation requires 'content' parameter."
+                )
+
+            handlers = {
+                "read": lambda: self._read_scratchpad(session_id),
+                "write": lambda: self._write_scratchpad(content, session_id),
+                "append": lambda: self._append_scratchpad(content, session_id),
+                "clear": lambda: self._clear_scratchpad(session_id),
+            }
+
+            # Operation is validated by Pydantic enum, so key existence is guaranteed
+            result = handlers[operation]()
+
+        return StringToolOutput(result=result)
+
+    def _raise_input_validation_error(self, message: str) -> None:
+        """Raise a ToolInputValidationError with the given message.
+
+        Args:
+            message: Error message to include in the exception.
+
+        Raises:
+            ToolInputValidationError: Always raised with the provided message.
+        """
+        raise ToolInputValidationError(message)
+
+    @classmethod
+    def get_scratchpad_for_session(cls, session_id: str) -> list[str]:
+        """Get scratchpad entries for a specific session.
+
+        Args:
+            session_id: Session identifier.
+
+        Returns:
+            List of scratchpad entries.
+        """
+        return cls._scratchpads.get(session_id, []).copy()
+
+    @classmethod
+    async def clear_session(cls, session_id: str) -> None:
+        """Clear scratchpad for a specific session.
+
+        Args:
+            session_id: Session identifier.
+        """
+        async with cls._lock:
+            if session_id in cls._scratchpads:
+                cls._scratchpads[session_id] = []
diff --git a/python/tests/tools/test_scratchpad.py b/python/tests/tools/test_scratchpad.py
new file mode 100644
index 000000000..55051b7b2
--- /dev/null
+++ b/python/tests/tools/test_scratchpad.py
@@ -0,0 +1,341 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from unittest.mock import Mock
+
+import pytest
+
+from beeai_framework.context import RunContext
+from beeai_framework.tools import StringToolOutput, ToolInputValidationError
+from beeai_framework.tools.scratchpad import ScratchpadInput, ScratchpadTool
+
+
+@pytest.fixture
+def mock_context() -> RunContext:
+    """Create a mock RunContext with a test run_id."""
+    context = Mock(spec=RunContext)
+    context.run_id = "test-run-123"
+    context.conversation_id = None
+    context.agent_id = None
+    return context
+
+
+@pytest.fixture
+def tool(mock_context: RunContext) -> ScratchpadTool:
+    """Create a fresh scratchpad tool instance for each test."""
+    tool_instance = ScratchpadTool()
+    # Initialize the session and clear any existing data
+    session_id = tool_instance._get_session_id(mock_context)
+    if session_id in ScratchpadTool._scratchpads:
+        ScratchpadTool._scratchpads[session_id] = []
+    return tool_instance
+
+
+"""
+Unit Tests
+"""
+
+
+@pytest.mark.asyncio
+async def test_read_empty_scratchpad(tool: ScratchpadTool) -> None:
+    """Test reading an empty scratchpad."""
+    result = await tool.run(input=ScratchpadInput(operation="read"))
+    assert isinstance(result, StringToolOutput)
+    assert "empty" in result.result.lower()
+
+
+@pytest.mark.asyncio
+async def test_write_to_scratchpad(tool: ScratchpadTool) -> None:
+    """Test writing an entry to the scratchpad."""
+    result = await tool.run(
+        input=ScratchpadInput(operation="write", content="Test action completed")
+    )
+    assert isinstance(result, StringToolOutput)
+    assert "added" in result.result.lower() or "updated" in result.result.lower()
+
+
+@pytest.mark.asyncio
+async def test_write_and_read(tool: ScratchpadTool) -> None:
+    """Test writing and then reading from the scratchpad."""
+    # Write an entry
+    await tool.run(input=ScratchpadInput(operation="write", content="First entry"))
+
+    # Read it back
+    result = await tool.run(input=ScratchpadInput(operation="read"))
+    assert isinstance(result, StringToolOutput)
+    assert "First entry" in result.result
+
+
+@pytest.mark.asyncio
+async def test_write_key_value_pairs(tool: ScratchpadTool) -> None:
+    """Test writing key-value pairs to the scratchpad."""
+    result = await tool.run(
+        input=ScratchpadInput(
+            operation="write", content="city: New York, date: 2025-01-27"
+        )
+    )
+    assert isinstance(result, StringToolOutput)
+
+    # Read it back
+    read_result = await tool.run(input=ScratchpadInput(operation="read"))
+    assert "city: New York" in read_result.result
+    assert "date: 2025-01-27" in read_result.result
+
+
+@pytest.mark.asyncio
+async def test_write_merge_key_value_pairs(tool: ScratchpadTool) -> None:
+    """Test that writing new key-value pairs merges with existing ones."""
+    # Write first set of pairs
+    await tool.run(input=ScratchpadInput(operation="write", content="city: Boston"))
+
+    # Write second set of pairs (should merge)
+    await tool.run(input=ScratchpadInput(operation="write", content="date: 2025-01-28"))
+
+    # Read it back
+    result = await tool.run(input=ScratchpadInput(operation="read"))
+    assert "city: Boston" in result.result
+    assert "date: 2025-01-28" in result.result
+
+
+@pytest.mark.asyncio
+async def test_write_key_value_with_comma_in_value(tool: ScratchpadTool) -> None:
+    """Test writing a key-value pair where the value contains a comma."""
+    await tool.run(
+        input=ScratchpadInput(
+            operation="write", content="item: milk, bread, eggs, priority: high"
+        )
+    )
+    read_result = await tool.run(input=ScratchpadInput(operation="read"))
+    assert "item: milk, bread, eggs" in read_result.result
+    assert "priority: high" in read_result.result
+
+
+@pytest.mark.asyncio
+async def test_write_key_with_hyphens(tool: ScratchpadTool) -> None:
+    """Test that keys with hyphens are correctly parsed."""
+    await tool.run(
+        input=ScratchpadInput(
+            operation="write", content="Content-Type: application/json, user-id: 12345"
+        )
+    )
+    read_result = await tool.run(input=ScratchpadInput(operation="read"))
+    assert "Content-Type: application/json" in read_result.result
+    assert "user-id: 12345" in read_result.result
+
+
+@pytest.mark.asyncio
+async def test_append_to_scratchpad(tool: ScratchpadTool) -> None:
+    """Test appending to the last entry."""
+    # Write initial entry
+    await tool.run(input=ScratchpadInput(operation="write", content="Initial entry"))
+
+    # Append to it
+    result = await tool.run(
+        input=ScratchpadInput(operation="append", content="- additional info")
+    )
+    assert isinstance(result, StringToolOutput)
+    assert "appended" in result.result.lower()
+
+    # Read it back
+    read_result = await tool.run(input=ScratchpadInput(operation="read"))
+    assert "Initial entry - additional info" in read_result.result
+
+
+@pytest.mark.asyncio
+async def test_append_without_entry_fails(tool: ScratchpadTool) -> None:
+    """Test that appending without an existing entry returns an error."""
+    result = await tool.run(
+        input=ScratchpadInput(operation="append", content="some content")
+    )
+    assert isinstance(result, StringToolOutput)
+    assert "no entry" in result.result.lower()
+
+
+@pytest.mark.asyncio
+async def test_clear_scratchpad(tool: ScratchpadTool) -> None:
+    """Test clearing the scratchpad."""
+    # Write some entries
+    await tool.run(input=ScratchpadInput(operation="write", content="Entry 1"))
+    await tool.run(input=ScratchpadInput(operation="write", content="Entry 2"))
+
+    # Clear the scratchpad
+    result = await tool.run(input=ScratchpadInput(operation="clear"))
+    assert isinstance(result, StringToolOutput)
+    assert "cleared" in result.result.lower()
+
+    # Verify it's empty
+    read_result = await tool.run(input=ScratchpadInput(operation="read"))
+    assert "empty" in read_result.result.lower()
+
+
+@pytest.mark.asyncio
+async def test_write_without_content_fails(tool: ScratchpadTool) -> None:
+    """Test that write operation without content raises validation error."""
+    with pytest.raises(ToolInputValidationError, match="content"):
+        await tool.run(input=ScratchpadInput(operation="write"))
+
+
+@pytest.mark.asyncio
+async def test_append_without_content_fails(tool: ScratchpadTool) -> None:
+    """Test that append operation without content raises validation error."""
+    # Add an entry first
+    await tool.run(input=ScratchpadInput(operation="write", content="Entry"))
+
+    # Try to append without content
+    with pytest.raises(ToolInputValidationError, match="content"):
+        await tool.run(input=ScratchpadInput(operation="append"))
+
+
+@pytest.mark.asyncio
+async def test_invalid_operation(tool: ScratchpadTool) -> None:
+    """Test that an invalid operation is handled properly."""
+    with pytest.raises(ToolInputValidationError):
+        await tool.run(input={"operation": "invalid_op"})
+
+
+@pytest.mark.asyncio
+async def test_get_scratchpad_for_session(tool: ScratchpadTool) -> None:
+    """Test the class method to get scratchpad for a session."""
+    # Write some data
+    await tool.run(input=ScratchpadInput(operation="write", content="Test entry"))
+
+    # Get the cached session ID from the tool instance
+    session_id = tool._cached_session_id
+
+    # Get scratchpad using class method
+    entries = ScratchpadTool.get_scratchpad_for_session(session_id)
+    assert len(entries) > 0
+    assert "Test entry" in entries[0]
+
+
+@pytest.mark.asyncio
+async def test_clear_session_class_method(tool: ScratchpadTool) -> None:
+    """Test the class method to clear a specific session."""
+    # Write some data
+    await tool.run(input=ScratchpadInput(operation="write", content="Test entry"))
+
+    # Get the cached session ID from the tool instance
+    session_id = tool._cached_session_id
+
+    # Clear using async class method
+    await ScratchpadTool.clear_session(session_id)
+
+    # Verify it's empty
+    entries = ScratchpadTool.get_scratchpad_for_session(session_id)
+    assert len(entries) == 0
+
+
+@pytest.mark.asyncio
+async def test_call_with_dict_input(tool: ScratchpadTool) -> None:
+    """Test calling the tool with dictionary input instead of model."""
+    result = await tool.run(input={"operation": "write", "content": "Dict entry"})
+    assert isinstance(result, StringToolOutput)
+
+    read_result = await tool.run(input={"operation": "read"})
+    assert "Dict entry" in read_result.result
+
+
+@pytest.mark.asyncio
+async def test_missing_operation_field(tool: ScratchpadTool) -> None:
+    """Test that missing operation field raises validation error."""
+    with pytest.raises(ToolInputValidationError):
+        await tool.run(input={})
+
+
+@pytest.mark.asyncio
+async def test_session_id_requires_context() -> None:
+    """Test that session ID extraction requires a valid context."""
+    tool_instance = ScratchpadTool()
+
+    # Should raise ToolInputValidationError when no context provided
+    with pytest.raises(ToolInputValidationError, match="requires RunContext"):
+        tool_instance._get_session_id(None)
+
+
+@pytest.mark.asyncio
+async def test_session_id_requires_valid_identifier() -> None:
+    """Test that session ID extraction requires a valid identifier in context."""
+    tool_instance = ScratchpadTool()
+
+    # Create a context with no valid identifiers
+    empty_context = Mock(spec=RunContext)
+    empty_context.run_id = None
+    empty_context.conversation_id = None
+    empty_context.agent_id = None
+
+    # Should raise ToolInputValidationError when no valid identifier found
+    with pytest.raises(ToolInputValidationError, match="None found in context"):
+        tool_instance._get_session_id(empty_context)
+
+
+@pytest.mark.asyncio
+async def test_session_id_caching(mock_context: RunContext) -> None:
+    """Test that session ID is cached after first extraction."""
+    tool_instance = ScratchpadTool()
+
+    # First call should extract and cache
+    session_id_1 = tool_instance._get_session_id(mock_context)
+    assert session_id_1 == "test-run-123"
+    assert tool_instance._cached_session_id == "test-run-123"
+
+    # Modify the context
+    mock_context.run_id = "different-run-456"
+
+    # Second call should return cached value, not re-extract
+    session_id_2 = tool_instance._get_session_id(mock_context)
+    assert session_id_2 == "test-run-123"  # Still the original cached value
+
+
+@pytest.mark.asyncio
+async def test_session_id_preference_order() -> None:
+    """Test that session ID extraction follows the correct preference order."""
+    tool_instance = ScratchpadTool()
+
+    # Test 1: run_id takes precedence
+    context1 = Mock(spec=RunContext)
+    context1.run_id = "run-123"
+    context1.conversation_id = "conv-456"
+    context1.agent_id = "agent-789"
+    assert tool_instance._get_session_id(context1) == "run-123"
+
+    # Test 2: conversation_id is second
+    tool_instance2 = ScratchpadTool()
+    context2 = Mock(spec=RunContext)
+    context2.run_id = None
+    context2.conversation_id = "conv-456"
+    context2.agent_id = "agent-789"
+    assert tool_instance2._get_session_id(context2) == "conv-456"
+
+    # Test 3: agent_id is last
+    tool_instance3 = ScratchpadTool()
+    context3 = Mock(spec=RunContext)
+    context3.run_id = None
+    context3.conversation_id = None
+    context3.agent_id = "agent-789"
+    assert tool_instance3._get_session_id(context3) == "agent-789"
+
+
+@pytest.mark.asyncio
+async def test_concurrent_writes_do_not_corrupt_state(tool: ScratchpadTool) -> None:
+    """Test that concurrent writes do not corrupt the scratchpad."""
+    import asyncio
+
+    async def write_entry(content: str) -> None:
+        await tool.run(input=ScratchpadInput(operation="write", content=content))
+
+    # With key-value merging, each write will update the value for 'entry'.
+    # We run them concurrently to test for race conditions.
+    tasks = [write_entry(f"entry: {i}") for i in range(10)]
+    await asyncio.gather(*tasks)
+
+    result = await tool.run(input=ScratchpadInput(operation="read"))
+
+    # The final state should contain one of the written values due to merging.
+    # Without proper locking, the final state could be unpredictable.
+    assert "entry:" in result.result
+    # Check that it's a single consolidated entry
+    assert result.result.count("entry:") == 1
+    # Verify the value is one of the expected outcomes from concurrent writes
+    # The exact value depends on the order of completion, but it should be one of the 'entry: i' values
+    found_value = result.result.split("entry: ")[1].split()[0]
+    assert found_value.isdigit() and 0 <= int(found_value) < 10