Concurrency

A single lock is the simplest solution — until a slow background job makes your user wait. Chapter 10 fixes this with named lanes that give every concern its own independent queue.

The Big Picture

Chapter 7 (Single-Agent Loop) used a single threading.Lock — it works, but all concerns (user, heartbeat, cron) compete for the same door.

Chapter 10 upgrades to named LaneQueues — each concern has its own lane.

User input
    │
    ▼
CommandQueue.dispatch()
    │
    ├──> Lane "main"      ← user messages, FIFO, priority
    ├──> Lane "heartbeat" ← background check, non-blocking
    └──> Lane "cron"      ← scheduled jobs, non-blocking
         │
    LaneQueue.pump()
         │
    Future → result back to caller

With a single lock, a slow cron job can block the user from getting a response. Named lanes solve this by giving each concern its own independent queue.

Analogy: An airport has separate lanes — priority boarding, fast track, regular. Each lane has its own queue and moves independently.

Four Key Concepts

1. LaneQueue — Data Structure

type Task struct {
    fn         func() (any, error)
    result     chan result
    generation int
}

type result struct {
    value any
    err   error
}

type LaneQueue struct {
    name        string
    mu          sync.Mutex
    cond        *sync.Cond
    queue       []Task
    activeCount int
    generation  int // ← restart recovery
}

func NewLaneQueue(name string) *LaneQueue {
    lq := &LaneQueue{name: name}
    lq.cond = sync.NewCond(&lq.mu)
    return lq
}

2. pump() — Self-Pumping Design

No external scheduler. Each LaneQueue manages itself:

func (cq *CommandQueue) pump(lane *LaneQueue) {
    for {
        lane.mu.Lock()
        for len(lane.queue) == 0 {
            lane.cond.Wait() // sleep when empty
        }
        task := lane.queue[0]
        lane.queue = lane.queue[1:]
        gen := lane.generation
        lane.mu.Unlock()

        if task.generation != gen {
            task.result <- result{err: errors.New("cancelled: stale generation")} // stale — skip
            continue
        }

        lane.mu.Lock()
        lane.activeCount++
        lane.mu.Unlock()

        val, err := task.fn()
        task.result <- result{value: val, err: err}

        lane.mu.Lock()
        lane.activeCount--
        lane.mu.Unlock()
    }
}

Thread sleeps when the queue is empty → wakes via condition.notify() when a task arrives. No polling, no busy-wait.

When a new task is dispatched to this lane, it calls cond.Signal() — this wakes the sleeping goroutine so it can pick up the task immediately.

3. Future-Based Results

In a chat bot, blocking means the UI freezes until the LLM responds. Futures let the system stay responsive while the request is in flight.

Callers don’t block directly — they get a Future and can wait whenever they’re ready:

// Dispatch task to lane
ch := commandQueue.Dispatch("main", func() (any, error) {
    return llmCall(message)
})

// Wait for result (blocking, with timeout)
select {
case res := <-ch:
    result, err = res.value, res.err
case <-time.After(30 * time.Second):
    err = errors.New("timeout")
}

// Or check without blocking
select {
case res := <-ch:
    result, err = res.value, res.err
default:
    // not ready yet
}

This pattern keeps the UI responsive while the LLM is thinking.

4. Generation Tracking — Restart Recovery

When an agent restarts (crash, redeploy), old tasks in the queue become stale:

func (cq *CommandQueue) RestartLane(laneName string) {
    lane := cq.lanes[laneName]
    lane.mu.Lock()
    defer lane.mu.Unlock()
    lane.generation++    // bump generation
    lane.queue = nil     // clear the queue
    lane.cond.Broadcast()
}

Tasks from the previous generation check task.generation != lane.generation → auto-cancelled. No zombie tasks running after a restart. The caller receives a cancellation error — so nothing hangs silently.

Chapter 7 (Single-Agent Loop) vs Chapter 10 (Concurrency)

Graceful Shutdown

A production agent should handle SIGTERM cleanly:

SIGTERM received
    │
    ├── stop accepting new tasks
    ├── drain active lanes (wait for activeCount → 0)
    ├── flush delivery-queue to disk (Chapter 8 (Delivery Guarantees) guarantee)
    └── exit cleanly

If the LLM is mid-response when the signal arrives, the write-ahead delivery queue from Chapter 8 (Delivery Guarantees) handles the pending message on the next startup. The two patterns compose: graceful shutdown + write-ahead delivery = no lost messages.

Takeaway

• Named lanes — each concern (user/heartbeat/cron) has its own queue — no blocking
• pump() self-pumping — no external scheduler needed
• Future pattern — caller gets async results without busy-waiting
• Generation tracking — stale tasks from before restart are auto-cancelled
• Graceful shutdown — drain lanes, flush queue — Chapter 8 (Delivery Guarantees) as safety net
• Chapter 10 — the recommended concurrency model once your agent has multiple concerns running in parallel

Cross-References

• Chapter 7 (Single-Agent Loop) — the single-lock pattern this chapter replaces
• Chapter 8 (Delivery Guarantees) — write-ahead queue that pairs with graceful shutdown
• Chapter 9 (Resilience) — retry chain that runs inside a lane task