Thread-Safe Agent State

Thread-safe chat history and cache management

PraisonAI protects chat history, caches, and session state when agents run from multiple threads or async tasks.

from praisonaiagents import Agent
import threading

agent = Agent(name="Assistant", instructions="You are helpful.")

def worker(prompt):
    agent.chat(prompt)

threads = [threading.Thread(target=worker, args=(f"Question {i}",)) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

The user drives one agent from several threads; locks keep chat history and caches consistent without corrupted state.

How It Works

Quick Start

1. Multi-threaded Chat

   from praisonaiagents import Agent
   import threading
   
   agent = Agent(name="ThreadSafeAgent", instructions="You are helpful.")
   
   def worker(prompt):
       response = agent.chat(prompt)
       print(f"Response: {response[:50]}...")
   
   threads = [threading.Thread(target=worker, args=(f"Question {i}",)) for i in range(5)]
   for t in threads:
       t.start()
   for t in threads:
       t.join()

2. Async Concurrent Tasks

   import asyncio
   from praisonaiagents import Agent
   
   agent = Agent(name="AsyncAgent")
   
   async def async_chat(prompt):
       return agent.chat(prompt)
   
   async def main():
       results = await asyncio.gather(*[async_chat(f"Question {i}") for i in range(5)])
   
   asyncio.run(main())

Thread-Safe Components

Chat History

The chat_history property is now fully thread-safe with automatic locking. The SDK protects chat history mutations through internal helper methods and a locked setter:

from praisonaiagents import Agent
import threading

agent = Agent(
    name="ThreadSafeAgent",
    instructions="You are helpful."
)

def worker(prompt):
    # Safe to call from multiple threads
    response = agent.chat(prompt)
    print(f"Response: {response[:50]}...")

# Create multiple threads
threads = [
    threading.Thread(target=worker, args=(f"Question {i}",))
    for i in range(5)
]

# Start all threads
for t in threads:
    t.start()

# Wait for completion
for t in threads:
    t.join()

What changed in PR #1488

What changed in PR #3769 — per-turn rollback ownership

Concurrent chat() / achat() turns on the same Agent used to interfere on rollback: if turn A failed after turn B had already appended new messages, A’s positional rollback (chat_history[:snapshot_len]) would erase B’s messages too.

_rollback_chat_history_to() now identifies messages by object identity and removes only the messages the failing turn actually appended. A concurrent turn’s messages are never touched.

The mechanism uses contextvars.ContextVar — each chat() / achat() turn runs in its own thread or task and gets an isolated ownership list. The append helpers (_add_to_chat_history, _add_to_chat_history_if_not_duplicate) record every message into the owning turn’s list; on failure, only those messages are removed.

Practical effect: you can safely run multiple concurrent chat() / achat() calls on the same Agent, and a failure on one turn (LLM error, guardrail rejection, response validation) no longer corrupts a concurrent turn’s chat history.

Follow-up in PR #4593. The ownership tracker described above only actually caught real turns after PR #4593. Before that, the tracker was fed only from _add_to_chat_history, but every chat() / achat() turn appends via _append_to_chat_history, so _rollback_chat_history_to always took the unsafe positional-delete branch and an empty-but-tracked turn also fell through (if owned: is falsy for []). PR #4593 wires _append_to_chat_history into the tracker and switches the rollback check to if owned is not None:, so concurrent-turn rollback safety now behaves as this section describes. Reference: PraisonAI PR #4593, fixes #4592.

Completed on the default (OpenAI) path in PR #5247. PR #3769’s ownership tracking was only wired into the custom-LLM branch; the default (OpenAI) chat() / achat() branches still fell through to the positional del chat_history[n:] rollback and could erase a concurrent turn’s messages. If your workflow shares one Agent across concurrent chat() / achat() turns on the default OpenAI path (the most common setup), you are now safe — a failing turn removes only its own messages, never a concurrent turn’s. Both branches now begin tracking before the turn: the sync branch ends it in a finally (after rollback reads the ownership list), and the async branch relies on the next turn’s _clear_turn_tracking(), matching the async custom-LLM branch. As a bonus, a terminal achat() / arun() / astart() no longer leaves its completed ownership list registered in the live-turn registry. Reference: PraisonAI PR #5247, closes #5149.

What changed in PR #5121 — OpenAI-native path TOCTOU parity

PR #3769 closed the read-then-append TOCTOU on the custom-LLM branches only. The OpenAI-native sync _chat_impl and async _achat_impl paths were still on the unsafe pattern: a concurrent turn’s user message that landed at chat_history[-1] between the read and the append was mistaken for this turn’s own duplicate, silently dropping this turn’s user message from the record.

PR #5121 routes both OpenAI-native paths through the same atomic _add_to_chat_history_if_not_duplicate("user", …) helper. Concurrent same-prompt turns on the OpenAI-native path no longer drop the user message.

All four dispatch branches now share the atomic helper:

No branch is left on the positional-rollback fallback — the default (OpenAI) path and the custom-LLM path now share the same rollback semantics.

What changed in PR #1514

What changed in PR #3769

Safe operations

# These operations are now thread-safe out of the box:
agent.chat_history = []  # Full replacement - uses locked setter
agent.chat("Hello")      # Appends safely via internal methods

# Reading is always safe:
history = agent.chat_history
print(f"History has {len(history)} messages")

Caches

Internal caches use threading.RLock for reentrant locking:

• _system_prompt_cache – Cached system prompts
• _formatted_tools_cache – Cached tool definitions

Rate Limiter

RateLimiter can be shared across threads and agents. Both the sync and async method families are fully locked — see Rate Limiter → Thread Safety & Multi-Agent Use for patterns.

Telemetry Counting

Telemetry counting is thread-safe as of PraisonAI #4235: the re-entrancy guard is a per-agent contextvars.ContextVar, so concurrent chat()/run()/start() calls on one shared agent are each counted independently. See Telemetry — Concurrent-execution accuracy.

Concurrent token/cost attribution on a shared LLM (PraisonAI #5233)

A single LLM instance can back two Agents (Agent(llm=shared_llm) on both). Under PraisonAIAgents.arun_all_tasks those tasks are dispatched via asyncio.gather on one thread — so before this fix, task A’s set_current_agent("Researcher") was overwritten by task B’s set_current_agent("Writer") on the same LLM object, and when A’s awaited response returned it read "Writer". A’s tokens got attributed to B in the collector’s by_agent map.

LLM.current_agent_name is now backed by a per-instance contextvars.ContextVar (self._current_agent_name_var), exposed via a backward-compatible property + set_current_agent() method. asyncio.gather copies the context per task at dispatch time, so each concurrently-gathered coroutine keeps its own value across the await. This guarantee was further extended in PR #5357.

Per-instance (not class-level) is deliberate: two distinct LLM instances — including an original and its copy.deepcopy clone — must keep independent attribution state within the same context, otherwise the last writer’s agent name would leak across LLMs. (Same reasoning as _tools_scope_depths in agents/agents.py.)

User impact: no API change. If you use the Agent(llm=shared_llm) pattern (single LLM object shared across a team), your per-agent by_agent totals in get_token_collector().get_session_summary() and PraisonAIAgents.get_token_usage_summary() are now correct under concurrent execution.

from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents.llm import LLM

# One LLM instance shared across two agents
shared_llm = LLM(model="gpt-4o-mini")

researcher = Agent(name="Researcher", llm=shared_llm, instructions="Research the topic")
writer     = Agent(name="Writer",     llm=shared_llm, instructions="Write from the research")

r = Task(name="research", description="Research quantum computing", agent=researcher, async_execution=True)
w = Task(name="write",    description="Write an article",           agent=writer,     async_execution=True, context=[r])

team = PraisonAIAgents(agents=[researcher, writer], tasks=[r, w])
team.start()  # arun_all_tasks under the hood — concurrent asyncio.gather

# by_agent now correctly attributes tokens to Researcher vs Writer
print(team.get_token_usage_summary()["by_agent"])

See also Token Tracking — Multi-agent breakdown.

LiteAgent Thread Safety

The lite package also provides thread-safe operations:

from praisonaiagents.lite import LiteAgent, create_openai_llm_fn
import threading

llm_fn = create_openai_llm_fn(model="gpt-4o-mini")
agent = LiteAgent(name="LiteThreadSafe", llm_fn=llm_fn)

def concurrent_chat(message):
    return agent.chat(message)

# Safe concurrent access
with threading.ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(concurrent_chat, f"Q{i}") for i in range(10)]
    results = [f.result() for f in futures]

Implementation Details

Lock Types

Deep-copy support

Agent.__deepcopy__ replaces threading.RLock (__cache_lock) and threading.Lock (_cost_lock) with fresh instances during copy.deepcopy(agent), so the agent is copy-safe on every supported Python version (RLock pickling was broken on CPython < 3.13).

Lock Usage Pattern

# Internal implementation pattern
class Agent:
    def __init__(self):
        self.__chat_history_state = AsyncSafeState([])
        self._cache_lock = threading.RLock()

    @property
    def _history_lock(self):
        return self.__chat_history_state

    def _append_to_chat_history(self, message):
        from .memory_mixin import _active_turn_owned
        with self._history_lock.lock():
            self._history_lock.value.append(message)
            owned = _active_turn_owned.get()
            if owned is not None:
                owned.append(message)

Why a re-entrant lock?

Nested calls (e.g. a helper that holds the lock and then assigns chat_history, which itself acquires the lock) used to deadlock. RLock permits the same thread to re-enter. See PR #1567 for details.

Persistence Orchestrator session cache

PersistenceOrchestrator guards its in-memory session cache with threading.RLock, returns deep copies on read, and (since PR #3792) bounds cache size via a collections.OrderedDict LRU. Size is capped by PRAISONAI_SESSION_CACHE_MAX (default 1024, minimum 1); empty/non-numeric values fall back to the default. Concurrent agents can share an orchestrator without corrupting cached ConversationSession objects and without unbounded memory growth in long-running servers.

The session cache stays orchestrator-owned, but after PR #4394 the orchestrator’s message-write and session-end store operations delegate to PraisonAIDB (through the internal _call_store / _dispatch_async helpers). The adapter’s “lazy store initialisation is race-free” guarantee therefore now covers both persistence entry points.

from praisonaiagents import Agent
from praisonai.persistence import PersistenceOrchestrator
import threading

# Create shared orchestrator
orchestrator = PersistenceOrchestrator()

def agent_worker(agent_name, session_id):
    agent = Agent(name=agent_name)

    # Safe: orchestrator returns deep copy of cached session
    history = orchestrator.on_agent_start(agent, session_id=session_id)

    # Multiple agents can safely read/update sessions concurrently
    orchestrator.on_message(session_id, "user", "Hello from " + agent_name)
    orchestrator.on_agent_end(agent, session_id)

# Multiple threads can safely share the orchestrator
threads = [
    threading.Thread(target=agent_worker, args=(f"Agent{i}", "session-123"))
    for i in range(3)
]

for t in threads:
    t.start()
for t in threads:
    t.join()

DefaultSessionStore — cross-process safety

DefaultSessionStore uses a per-session file lock (fcntl.flock on Unix, msvcrt.locking on Windows) for every write path:

Reads inside the lock guarantee that a metadata merge picks up any messages another worker appended just before. Since PR #1709, FileLock.__enter__ raises IOError on timeout (previously the lock failure was silent, which could lead to torn writes on a stuck lock). The default lock_timeout is 5.0 seconds — pass DefaultSessionStore(lock_timeout=...) to tune it.

Cross-session isolation also covers id collisions on disk: distinct session ids always map to distinct files, and a file whose stored session_id does not match is never served (PraisonAI PR #6061). See On-disk filenames.

storage.base.FileLock — mtime staleness + heartbeat

praisonaiagents.storage.base.FileLock backs BaseJSONStore and every JSON-backed store built on it (including the learned-memory store). It judges staleness by the lock file’s mtime age, not by how long any individual waiter has queued.

This closes a corruption bug: previously, staleness was measured as time.time() - start, where start was when this waiter began waiting. A live holder legitimately in a long critical section (large JSON write, slow disk) had its lock file unlink()ed out from under it as soon as one waiter’s timeout elapsed — admitting two writers into the same read-modify-write save concurrently and silently clobbering one another’s writes. The mtime + heartbeat scheme keeps a long-held lock exclusive. Added in PR #5136.

BaseJSONStore — atomic save even on serialization failure

BaseJSONStore._save writes to a .tmp staging file next to the store and os.replaces it over the real file so partial content is never observable. As of PR #5582, the atomic guarantee extends to failures inside json.dump itself — a circular dict, a UTF-8-unencodable surrogate string, or any other serialization or encoding error.

Previously, temp_path = f.name was assigned after json.dump succeeded, so a serialization failure left temp_path unbound and the existing cleanup branch couldn’t identify the staging file. Both locked and unlocked save paths were affected. Fix reference: PraisonAI PR #5582, fixes #5580.

sequenceDiagram
    participant Caller
    participant Store as BaseJSONStore
    participant Tmp as session.json.tmp
    participant Real as session.json

    Caller->>Store: save(data_with_circular_ref)
    Store->>Tmp: create NamedTemporaryFile
    Store->>Store: temp_path = f.name  (recorded first)
    Store->>Tmp: json.dump(data, …)
    Tmp-->>Store: ValueError: Circular reference
    Store->>Tmp: os.unlink(temp_path)   ← cleanup now identifies the file
    Store-->>Caller: re-raise original error
    Note over Real: untouched — no partial write ever reached it

    classDef caller fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef store fill:#189AB4,stroke:#7C90A0,color:#fff
    classDef tmp fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef real fill:#10B981,stroke:#7C90A0,color:#fff

    class Caller caller
    class Store store
    class Tmp tmp
    class Real real

BaseJSONStore — read failures block ordinary saves on the same instance

BaseJSONStore and AsyncBaseJSONStore share a per-instance read-failure guard: when _load() / load_async() catches an UnicodeDecodeError, a json.JSONDecodeError, or a non-missing IOError, the instance returns defaults and sets _read_failed = True. A subsequent save() / save_async() then raises OSError: Refusing to overwrite unreadable record: <path> for as long as the source file still exists. Neighbouring stores and other keys on a shared backend are unaffected — the guard is per-instance / per-path.

Clearing the guard requires one of these:

sequenceDiagram
    participant Caller
    participant Store as BaseJSONStore
    participant Disk as session.json (bad bytes)

    Caller->>Store: load()
    Store->>Disk: open + json.load
    Disk-->>Store: UnicodeDecodeError
    Store-->>Caller: {} (defaults)
    Note over Store: _read_failed = True

    Caller->>Store: save({"ok": 1})
    Store->>Disk: exists? yes
    Store-->>Caller: OSError<br/>Refusing to overwrite<br/>unreadable record

    Caller->>Store: clear()   OR   delete()   OR   fix bytes + load()
    Note over Store: _read_failed = False
    Caller->>Store: save({"ok": 1})
    Store->>Disk: tmp + os.replace
    Disk-->>Store: ok

    classDef caller fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef store fill:#189AB4,stroke:#7C90A0,color:#fff
    classDef disk fill:#F59E0B,stroke:#7C90A0,color:#fff

    class Caller caller
    class Store store
    class Disk disk

Backend-delegated stores: when BaseJSONStore is constructed with a FileBackend (BaseJSONStore(path, backend=FileBackend(...))), the per-path guard lives on the backend, not the store. The contract is the same from the caller’s point of view: the first failed load() makes further save() calls on that key refuse to overwrite the file until the key is deleted, successfully reloaded, or the file is removed out-of-band.

Cross-instance / cross-process: no guarantee is claimed. A fresh instance pointed at the same bad file goes through the same recovery on its first load().

Fix references: PraisonAI PR #5578, PR #5579, fixes #5577.

SqliteTranscriptStore — cross-process safety

SqliteTranscriptStore is the default gateway transcript backend as of PraisonAI PR #3409. It replaces the FileLock above with a SQLite transaction: every read-modify-write runs its SELECT + INSERT OR REPLACE inside a single BEGIN IMMEDIATE transaction, which grabs the database RESERVED write lock and holds it until COMMIT.

That closes the multi-gateway lost-append window the process-local lock alone could not: a second gateway process cannot slip a read between this one’s read and its write. Gateway deployments with session.persist: true use this path by default — see SQLite Transcript Store.

Best Practices

response = agent.chat("Hello")

# Bad — bypasses locks
agent.chat_history.append({"role": "user", "content": "Hello"})

# Good
agent.chat("Hello")

Async Considerations

agent.chat_history is async-aware out of the box — no external asyncio.Lock is required when all calls are inside the same event loop.

import asyncio
from praisonaiagents import Agent

agent = Agent(name="AsyncAgent")

async def async_chat(prompt):
    # No external lock needed - AsyncSafeState handles this
    return agent.chat(prompt)

async def main():
    tasks = [async_chat(f"Question {i}") for i in range(5)]
    results = await asyncio.gather(*tasks)

An external lock is still useful for serialising chat-history mutations from a thread pool that mixes sync and async callers:

import asyncio
import threading
from praisonaiagents import Agent

agent = Agent(name="MixedAgent")
external_lock = threading.Lock()

def sync_worker(prompt):
    with external_lock:
        return agent.chat(prompt)

async def async_worker(prompt):
    # Convert to sync context for coordination
    loop = asyncio.get_event_loop()
    with external_lock:
        return await loop.run_in_executor(None, agent.chat, prompt)

Verifying Thread Safety

Test thread safety with concurrent access:

import threading
from praisonaiagents.lite import LiteAgent

def test_thread_safety():
    agent = LiteAgent(
        name="Test",
        llm_fn=lambda m: "Response"
    )

    errors = []

    def worker():
        try:
            for _ in range(100):
                agent.chat("Test")
        except Exception as e:
            errors.append(e)

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

    assert len(errors) == 0, f"Thread safety errors: {errors}"
    print("Thread safety test passed!")

test_thread_safety()

Multi-team HTTP launch

PraisonAI’s launch server is now fully thread-safe and unified across Agent.launch(), AgentTeam.launch(), and PraisonAIAgents.launch(). All three share one process-wide _server_registry in praisonaiagents/agent/server_registry.py, so mixing them on the same port is safe.

• Multiple Agent / Agents instances may call .launch(port=N) concurrently from different threads — registration is atomic.
• If two launch calls use the same path on the same port, the second gets an auto-suffixed path (/path_abc123) and a warning is logged.
• Server readiness is signalled deterministically (no fixed sleep); .launch() returns only after the port is accepting connections. The wait defaults to 5 seconds and is configurable via the PRAISONAI_SERVER_READY_TIMEOUT environment variable. If the server doesn’t become ready in time, .launch() still returns and a warning is logged — check server logs for startup errors.
• A bind failure (Address already in use, missing FastAPI/uvicorn) clears the port’s _started flag so a later .launch() call re-attempts the bind. The launcher returns the true bound state, so callers can distinguish a live server from a poisoned port.
• aworkflow() state lock is created inside the running async context, so workflows remain stable when invoked under pytest-asyncio or when nested inside another loop.
• Per-turn tool tracking (_turn_tools_used) is now protected by the same lock that guards chat_history, so concurrent chat() / achat() turns on one Agent no longer corrupt hook or self-improve tool data. No new public API — the buffer stays private; the read/append helpers were made async-safe internally.

import threading
from praisonaiagents import AgentTeam

def launch_team(team_name, port, path):
    team = AgentTeam(name=team_name)
    team.launch(port=port, path=path)

# Safe concurrent launches
thread1 = threading.Thread(target=launch_team, args=("TeamA", 8000, "/team_a"))
thread2 = threading.Thread(target=launch_team, args=("TeamB", 8000, "/team_b"))

thread1.start()
thread2.start()

thread1.join()
thread2.join()

# Both teams available at:
# - http://localhost:8000/team_a
# - http://localhost:8000/team_b

Wrapper-layer thread safety (praisonai package)

The praisonai wrapper layer (distinct from the praisonaiagents content above) provides thread-safe OpenAI client management and CLI command discovery.

Per-instance OpenAI client lifecycle

Each BaseAutoGenerator owns its own core OpenAIClient (from praisonaiagents.llm.openai_client), which manages sync and async access internally — no cross-instance sharing, no LRU eviction surprises.

OpenAIClient is also loop-aware: concurrent sync callers each on their own asyncio.run loop each get an AsyncOpenAI bound to their own loop, and the drop/rebuild is serialised under an internal threading.Lock, so no caller ever observes a transient None. Added in PraisonAI PR #5682.

import os
from praisonai import PraisonAI

# Each PraisonAI instance constructs its own auto-generator, which owns
# a single OpenAI client. Two concurrent instances cannot evict each
# other's client — even with identical API keys.
team_a = PraisonAI(auto="Draft a marketing plan", api_key=os.getenv("OPENAI_API_KEY"))
team_b = PraisonAI(auto="Draft a technical doc", api_key=os.getenv("OPENAI_API_KEY"))

team_a.run()  # safe to run in parallel threads
team_b.run()

For power users building generators directly:

import os
from praisonai.auto import BaseAutoGenerator

# Sync context manager
with BaseAutoGenerator(config_list=[{
    "model": "gpt-4o-mini",
    "api_key": os.getenv("OPENAI_API_KEY"),
    "base_url": None,
}]) as gen:
    # _get_core_client() is lock-guarded; calling it from multiple threads
    # on the same instance returns the same core OpenAIClient.
    result = gen._structured_completion(MyModel, messages=[...])

# Async context manager
async with BaseAutoGenerator(config_list=[{
    "model": "gpt-4o-mini",
    "api_key": os.getenv("OPENAI_API_KEY"),
    "base_url": None,
}]) as gen:
    result = await gen._structured_completion_async(MyModel, messages=[...], is_async=True)

Thread-safe Typer command discovery

Embedding python -m praisonai from multiple threads is now safe. The CLI command discovery uses a double-check lock pattern and doesn’t poison the cache on failure:

import threading
import subprocess

def run_cli_command(command):
    # Safe to call from multiple threads
    result = subprocess.run(
        ["python", "-m", "praisonai"] + command,
        capture_output=True, text=True
    )
    return result.stdout

# Multiple threads can safely use the CLI
threads = [
    threading.Thread(target=run_cli_command, args=(["--version"],))
    for _ in range(5)
]

for t in threads:
    t.start()
for t in threads:
    t.join()

Failure-safe cache

A transient discovery error does not lock the CLI into a broken state — the next call retries instead of permanently breaking dispatch. This ensures reliable operation in multi-threaded server environments where temporary import failures might occur.

New Thread-Safe Components in PR #1548

AsyncAgentScheduler is now loop-aware. The start() method binds its async primitives (asyncio.Event, asyncio.Lock) to the running loop, and stop() raises RuntimeError if called from a different loop than start().

Lazy loaders in praisonai/auto.py are now thread-safe. A single _load_optional(key, loader) helper with a module-level lock replaces the previous unguarded module-level globals.

inbuilt_tools lazy import (PR #1681) now routes through praisonai.auto._load_optional("inbuilt_autogen_tools", ...) instead of a hand-rolled re-entry guard. Negative results are cached, so a missing crewai or autogen install no longer pays the find_spec cost on every attribute access.

Framework availability constants (PR #1780) — Module-level constants on praisonai.agents_generator (AGENTOPS_AVAILABLE, etc.) are resolved lazily via __getattr__. In praisonai.observability.hooks, the eager AGENTOPS_AVAILABLE constant was removed (PR #2062) — use is_agentops_available() instead.

Integration registry (praisonai/integrations/registry.py) now has a per-instance threading.Lock guarding register/unregister/create/list_registered operations.

PraisonAIDB background writes (PR #4861) — fire-and-forget write tracking (_bg_writes / _bg_writes_lock) is now per-instance, not a process-global set. Each adapter’s close() / aclose() / flush_pending_writes() waits only on its own in-flight writes. Previously a single process-global _BG_WRITES was shared across every PraisonAIDB instance, so one tenant’s stuck backend could consume another tenant’s 5-second flush budget — or silently discard the caller’s own writes when it timed out. The per-instance lock also removes unnecessary serialisation across concurrent multi-tenant close() calls. See Async DB Hooks.

The adapter’s async init/close is also cross-loop safe: a shared PraisonAIDB no longer caches an event-loop-bound asyncio.Lock on the instance, so its _ainit_stores / aclose can be awaited from any per-session loop (serialisation runs off-loop via asyncio.to_thread under the sync threading.Lock) without RuntimeError: bound to a different event loop. See Per-Session Scoped Bridge.

New Thread-Safe Components in PR #1673

Jobs server singleton init (PR #1771) — get_store() and get_executor() in praisonai/jobs/server.py now use double-checked locking with threading.Lock. This eliminates a TOCTOU race where concurrent cold-start requests could create orphaned JobStore / JobExecutor instances on a fresh process.

AgentTeam / Workflow run guard (PR #5656) — Both AgentTeam and Workflow lazily mint their per-instance run lock through a single shared helper using double-checked locking under a module-level guard, so two threads racing into the first start() / astart() / run() on a fresh instance observe the same lock object.

InMemoryJobStore — locked reads and async get_stats()

All read methods (get, get_by_idempotency_key, list_jobs, count, get_stats) now hold an asyncio.Lock while reading internal dicts, so concurrent saves cannot tear a read.

AgentScheduler — interruptible retry backoff (sync scheduler)

stop() now becomes responsive within milliseconds even during retry backoff. The sync scheduler also adopts the shared backoff_delay() curve so sync and async retries are identical.

from praisonai.scheduler import AgentScheduler

scheduler = AgentScheduler(agent, "Hourly news check")
scheduler.start("hourly", max_retries=3)
# ... later, from another thread or signal handler:
scheduler.stop()  # returns within ms, even if a backoff sleep was in progress

ToolRegistry — thread-safe registry operations

ToolRegistry now holds a threading.Lock around all reads and mutations, matching PluginRegistry / integration registry. Eliminates RuntimeError: dictionary changed size during iteration when registering tools concurrently with iteration. Reference: PR #1673.

MCP registry lazy-loader locks (PR #2738)

All three MCP registries (MCPToolRegistry, MCPResourceRegistry, MCPPromptRegistry) now guard the lazy-loader queue with a threading.Lock using an atomic snapshot-then-run pattern. Concurrent list_all / get calls from stdio and HTTP transports are safe — each loader runs exactly once, regardless of caller count.

Each _ensure_loaded acquires the lock, snapshots _lazy_loaders, clears the pending set, releases the lock, then runs each loader outside the critical section:

Under 10-thread concurrency, loaders can no longer double-run or be skipped mid-iteration. Reference: PR #2738.

PluginManager (PR #5204)

PluginManager‘s “thread-safe for concurrent access” contract is now fully honoured. Every method that iterates or mutates _plugins / _enabled acquires the same RLock:

Concurrent register() / unregister() on one thread and execute_hook() on another can no longer raise RuntimeError: dictionary changed size during iteration. Because _lock is an RLock, a plugin whose on_shutdown() re-enters the manager (e.g. via shutdown()) is safe.

Fix reference: PraisonAI PR #5204.

Managed agents (PR #5230)

Managed agents — AnthropicManagedAgent / LocalManagedAgent guard cloud-resource creation and local session persistence with per-instance locks; concurrent execute() calls on a shared instance never double-create. Agent / environment / session creation uses double-checked locking, so only one of any resource is ever created — no orphaned cloud agents, no torn on-disk session JSON.

See Managed Agent Lifecycle → Thread safety for a runnable concurrent-execute() example.

Fix reference: PraisonAI PR #5230.

Multi-agent context safety (PR #1723)

praisonai.cli.features.interactive_tools._get_shared_runtime() and praisonai.tool_resolver._get_default_resolver() were previously process-wide singletons. They are now per-context via contextvars.ContextVar:

• Concurrent agents in the same process no longer share an InteractiveRuntime (no LSP/ACP config bleed).
• Resolvers anchor to each agent’s CWD, not whichever agent ran first.
• For long-lived daemons that switch projects, call praisonai.tool_resolver.reset_default_resolver() to force re-anchoring.

This eliminated the singleton-config-bleed and first-caller-CWD-wins bugs while keeping cleanup_runtime() and convenience functions like resolve_tool() source-compatible.

User tool loading is safe under concurrent builds too: _safe_loader.load_user_module registers each load under a per-load-unique sys.modules key, so two tenants loading their own tools.py never share the process-global slot. The entry is dropped after exec on both success and exception, so a long-lived multi-tenant praisonai serve process does not accumulate one sys.modules entry per reload — a failed exec still removes only its own entry, never another tenant’s live module. See Concurrent loads stay isolated. Fix references: PraisonAI PR #5248 + #5239 (closes #5157 Gap 3).

Handoff history seeding

A handoff seeds the filtered conversation onto the target agent’s chat_history for the duration of the transfer, then restores it. Because that history is shared mutable state on the target, overlapping handoffs to the same target agent (e.g. two parallel_handoffs entries pointing at one specialist) would interleave their seed-and-restore steps and leak one caller’s context into another’s.

Seeding is now serialised per target agent by a DualLock attached lazily as agent._handoff_seed_lock:

• Sync path holds the lock’s threading (RLock) side around apply-seed → chat() → restore.
• Async path holds the lock’s per-loop asyncio.Lock side and the sync side for the whole seeded achat() turn. The async side alone would not exclude a concurrent cross-thread sync handoff to the same target, so both sides are held together.

If a target agent cannot hold the attribute (e.g. __slots__), a transient lock is used and behaviour falls back to the pre-lock path — the common case stays fully guarded. No public API changed; the guarantee is automatic. See Agent Handoffs and Multi-Agent Context Safety.

Handoff chain isolation under asyncio.gather

Every task spawned by parallel_handoffs (and any asyncio.gather over handoffs) gets its own handoff chain — sibling tasks cannot corrupt each other’s cycle-detection or max-depth state.

import asyncio
from praisonaiagents import Agent
from praisonaiagents import parallel_handoffs

billing = Agent(name="Billing", instructions="Handle billing inquiries.")
refund = Agent(name="Refunds", instructions="Process refund requests.")
tech = Agent(name="TechSupport", instructions="Solve technical problems.")
triage = Agent(name="Triage", instructions="Route inquiries to specialists.")

async def main():
    results = await parallel_handoffs(
        triage,
        targets=[
            (billing, "Why was I charged twice?"),
            (refund, "I want a refund for order #42"),
            (tech, "The app crashes on upload"),
        ],
    )
    for r in results:
        print(r.target_agent, r.success)

asyncio.run(main())

Each gathered task rebinds _handoff_chain_var to a fresh per-task copy on entry (_handoff_chain_var.set(list(_handoff_chain_var.get() or []))), which — on top of the existing copy-on-write push/pop — keeps sibling handoffs fully isolated. See parallel_handoffs reference for parameters.

• Safety-rejected handoffs are chain-neutral: a handoff blocked by a safety check never consumes a chain slot, so the parent’s cycle and max_depth counters stay accurate for its remaining handoffs in the same turn.

Output singleton (trace/status) lock (PR #5219)

output="status" and output="trace" route through a shared per-process singleton in praisonaiagents.output.status / praisonaiagents.output.trace. Both modules declared a _output_lock that, before this fix, was never acquired.

The lock is now actually acquired around every read-modify-write on the singleton’s shared state:

• the timing dicts _llm_start_time, _tool_start_times, and _agent_start_times,
• the call counter _ai_call_count,
• the session accumulators _session_tokens_in, _session_tokens_out, _session_cost, and _session_llm_calls.

Concrete before/after: an asyncio.gather over two agent.achat(...)s used to let one agent’s llm_end compute its duration from the other agent’s _llm_start_time (or crash on None once the first llm_end reset it), and the session token summary silently under-counted because += on shared counters raced. After the fix, both the per-call timings routed through the singleton and the aggregate token/cost totals stay consistent under concurrency.

import asyncio
from praisonaiagents import Agent

# Distinct names → accurate per-agent status/trace attribution (see the note below).
writer = Agent(name="Writer", instructions="Write a blurb.", output="status")
summarizer = Agent(name="Summarizer", instructions="Summarise notes.", output="status")

async def main():
    # The shared status singleton's metrics no longer corrupt under gather.
    await asyncio.gather(
        writer.achat("Blurb for the new dashboard"),
        summarizer.achat("Summarise v2.0 release notes"),
    )

asyncio.run(main())

Cross-links: Output and Output Styles.

Related