Thread Safety CLI

CLI commands for verifying thread-safe agent operations

Thread Safety CLI

Commands for testing and verifying thread-safe operations in PraisonAI Agents.

Verification Commands

Run Thread Safety Tests

Run the built-in thread safety tests:

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

Output:

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

Quick Thread Safety Check

Verify thread safety with a quick Python command:

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')
"

Lite Agent Thread Safety Check

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')
"

Testing Concurrent Access

With Real API

Test concurrent API calls (requires OPENAI_API_KEY):

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')
"

Lock Verification

Check Agent Locks

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')
"

Check LiteAgent Locks

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')
"

Verify re-entrant locking

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')
"

Verify async lock requires an event loop

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}')
"

CI/CD Integration

Add thread safety checks to your CI pipeline:

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

Related

• Thread Safety (Code)
• Lite Package
• Performance CLI