cd /path/to/praisonai-agents
python -m pytest tests/unit/test_thread_safety.py -v

tests/unit/test_thread_safety.py::TestLiteAgentThreadSafety::test_concurrent_chat_history_access PASSED
tests/unit/test_thread_safety.py::TestLiteAgentThreadSafety::test_concurrent_clear_history PASSED
tests/unit/test_thread_safety.py::TestAgentThreadSafety::test_agent_has_history_lock PASSED
tests/unit/test_thread_safety.py::TestAgentThreadSafety::test_agent_has_cache_lock PASSED

python -c "
from praisonaiagents import Agent
import threading

agent = Agent(name='Test', instructions='You are a helpful assistant.')

# Verify locks exist
assert hasattr(agent, '_history_lock'), 'Missing history lock'
assert hasattr(agent, '_cache_lock'), 'Missing cache lock'
assert agent._history_lock.is_async_context() is False, 'Should be sync context'

print('Thread safety locks: OK')
"

python -c "
from praisonaiagents.lite import LiteAgent
import threading

agent = LiteAgent(name='Test', llm_fn=lambda m: 'Response')

# Verify lock exists
assert hasattr(agent, '_lock'), 'Missing lock'

# Test concurrent access
errors = []
def worker():
    try:
        for _ in range(50):
            agent.chat('Test')
    except Exception as e:
        errors.append(e)

threads = [threading.Thread(target=worker) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

assert len(errors) == 0, f'Errors: {errors}'
print('LiteAgent thread safety: OK')
"

export OPENAI_API_KEY="your-key"
python -c "
from praisonaiagents import Agent
import threading
import os

if not os.environ.get('OPENAI_API_KEY'):
    print('Skipping: OPENAI_API_KEY not set')
    exit(0)

agent = Agent(
    name='ConcurrentTest',
    instructions='Reply with one word only.',
    llm='gpt-4o-mini',
    output="silent"
)

results = []
lock = threading.Lock()

def worker(i):
    response = agent.chat(f'Say hello {i}')
    with lock:
        results.append(response)

threads = [threading.Thread(target=worker, args=(i,)) for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f'Completed {len(results)} concurrent requests')
print('Concurrent access test: OK')
"

python -c "
from praisonaiagents import Agent
from praisonaiagents.agent.async_safety import AsyncSafeState
import threading

agent = Agent(name='Test', instructions='You are a helpful assistant.')

# Check lock types
assert isinstance(agent._history_lock, AsyncSafeState)
assert isinstance(agent._cache_lock, type(threading.RLock()))

print('Lock types verified: AsyncSafeState (DualLock with RLock) + RLock')
"

python -c "
from praisonaiagents.lite import LiteAgent
import threading

agent = LiteAgent(name='Test', llm_fn=lambda m: 'ok')

assert hasattr(agent, '_lock')
print('LiteAgent lock verified')
"

python -c "
from praisonaiagents.agent.async_safety import DualLock
lock = DualLock()
with lock.sync():
    with lock.sync():  # would deadlock with a plain threading.Lock
        print('Re-entrant sync acquire: OK')
"

python -c "
import asyncio
from praisonaiagents.agent.async_safety import DualLock

lock = DualLock()

async def ok():
    async with lock.async_lock():
        print('Async acquire inside event loop: OK')
asyncio.run(ok())

try:
    # Outside an event loop — must raise
    lock._get_async_lock()
except RuntimeError as e:
    print(f'Outside event loop correctly raised: {e}')
"

# .github/workflows/test.yml
- name: Thread Safety Tests
  run: |
    pip install praisonaiagents pytest
    python -m pytest tests/unit/test_thread_safety.py -v