diff --git a/src/timmy/agentic_loop.py b/src/timmy/agentic_loop.py
index 90f2124..a59cfa0 100644
--- a/src/timmy/agentic_loop.py
+++ b/src/timmy/agentic_loop.py
@@ -18,6 +18,7 @@ from __future__ import annotations
 import asyncio
 import logging
 import re
+import threading
 import time
 import uuid
 from collections.abc import Callable
@@ -59,6 +60,7 @@ class AgenticResult:
 # ---------------------------------------------------------------------------
 
 _loop_agent = None
+_loop_agent_lock = threading.Lock()
 
 
 def _get_loop_agent():
@@ -66,12 +68,18 @@ def _get_loop_agent():
 
     Returns the same type of agent as `create_timmy()` but with a
     dedicated session so it doesn't pollute the main chat history.
+
+    Thread-safe: uses a lock to prevent duplicate agent creation
+    when multiple loops start concurrently.
     """
     global _loop_agent
-    if _loop_agent is None:
-        from timmy.agent import create_timmy
+    if _loop_agent is not None:
+        return _loop_agent
+    with _loop_agent_lock:
+        if _loop_agent is None:
+            from timmy.agent import create_timmy
 
-        _loop_agent = create_timmy()
+            _loop_agent = create_timmy()
     return _loop_agent
 
 
diff --git a/tests/unit/test_agentic_loop.py b/tests/unit/test_agentic_loop.py
index 7bebd84..e860e8f 100644
--- a/tests/unit/test_agentic_loop.py
+++ b/tests/unit/test_agentic_loop.py
@@ -104,6 +104,46 @@ class TestGetLoopAgent:
         finally:
             al._loop_agent = saved
 
+    def test_thread_safe_creation(self):
+        """Concurrent calls must only create one agent (thread-safety)."""
+        import threading
+
+        import timmy.agentic_loop as al
+
+        saved = al._loop_agent
+        try:
+            al._loop_agent = None
+            mock_agent = MagicMock()
+            call_count = 0
+            barrier = threading.Barrier(4)
+
+            original_create = MagicMock(return_value=mock_agent)
+
+            def slow_create():
+                nonlocal call_count
+                call_count += 1
+                return original_create()
+
+            results = [None] * 4
+
+            def worker(idx):
+                barrier.wait()
+                results[idx] = al._get_loop_agent()
+
+            with patch("timmy.agent.create_timmy", side_effect=slow_create):
+                threads = [threading.Thread(target=worker, args=(i,)) for i in range(4)]
+                for t in threads:
+                    t.start()
+                for t in threads:
+                    t.join()
+
+            # All threads got the same agent
+            assert all(r is mock_agent for r in results)
+            # create_timmy called exactly once
+            assert call_count == 1
+        finally:
+            al._loop_agent = saved
+
 
 # ── _broadcast_progress ──────────────────────────────────────────────────────