Concurrency: Per-session locks and cron heartbeats

Before the code: from one assistant to a small team

Two real-world problems show up the moment you actually deploy this. A user might tap send twice while the assistant is still thinking, and the two threads will trample each other on the same session file. Or you may want the assistant to do work on its own schedule (a morning check-in, an end-of-day review) without a human starting the chat. Then the bigger question: instead of one big do-everything assistant, can you have a tiny team of specialists who share the same memory but each focus on one job? This module solves all three: per-session locks, scheduled heartbeats, and prefix-routed sub-agents.

One front desk, a few specialists, shared notes

Your message hits a small router. The router picks the right sub-agent based on a prefix. All sub-agents share the same memory store.

Quiz: Quiz

Loading practice…

Two new wrinkles show up in any real deployment. A user might double-tap a message while the agent is mid-thought. A scheduled task might fire at the same moment the user sends a message. Both can corrupt the session file if we let them. The fix is one lock per user, and a separate session id for cron work.

09-concurrency-scheduling/bot.py
python
from collections import defaultdict
import threading

# Per-session locking
session_locks: dict[str, threading.Lock] = defaultdict(threading.Lock)

One lock per user_id, created lazily on first access. defaultdict means no manual setup, just attribute access.

09-concurrency-scheduling/bot.py
python
def run_agent_turn(user_id: str, user_text: str) -> str:
    """Run agent turn with per-session lock to prevent race conditions."""
    with session_locks[user_id]:
        history = load_session(user_id)
        history.append({"role": "user", "content": user_text})

        tokens = estimate_tokens(history)
        if tokens > TOKEN_THRESHOLD:
            history = compact_session(history)
            save_session(user_id, history)

        # ... usual while-True loop

The whole agent loop is wrapped in the lock. Two simultaneous turns for the same user will serialise. Different users still run in parallel.

No, because the lock is per user_id, not global. Two different users hold two different locks. Only same-user concurrent turns serialise, which is exactly what we want, because they share a session file. Now for the heartbeat side. We use the third-party schedule library (already installed by make install): you register jobs with schedule.every and a daemon thread calls schedule.run_pending in a loop.

09-concurrency-scheduling/bot.py
python
import schedule


def heartbeat_task():
    """Example heartbeat: runs as an isolated cron session."""
    task_name = "daily-summary"
    user_id = f"cron:{task_name}"
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")

    try:
        reply = run_agent_turn(
            user_id,
            f"It is {timestamp}. Check memory for any pending tasks "
            f"or reminders, and write a brief status summary to memory."
        )
        print(f"  [cron] {reply[:200]}")
    except Exception as e:
        print(f"  [cron] Error: {e}")


def setup_heartbeats():
    schedule.every(24).hours.do(heartbeat_task)


def run_scheduler():
    """Daemon thread that runs the schedule loop."""
    while True:
        schedule.run_pending()
        time.sleep(60)

A cron task is just another caller of run_agent_turn. It uses a special user_id prefixed with cron: so it gets its own session and its own lock.

The agent does not know it is being called from cron. It just sees a user_id and a message. The cron: prefix gives the heartbeat its own session, so summary logs do not contaminate user conversations.

Quiz: Quiz

Loading practice…