Wait Tracking

Track why agents are waiting

WaitTracker explains why agents are idle. Essential for debugging stuck workflows.

Basic Usage

from splinter.coordination import WaitTracker, WaitReason

tracker = WaitTracker()

# Agent starts waiting
tracker.start_waiting("writer", WaitReason.WAITING_FOR_INPUT, source="researcher")

# Check wait status
status = tracker.get_status("writer")
print(status.reason)  # WAITING_FOR_INPUT
print(status.source)  # "researcher"

# Agent stops waiting
tracker.stop_waiting("writer")

Wait Reasons

class WaitReason(Enum):
    WAITING_FOR_INPUT = "waiting_for_input"      # Another agent's output
    WAITING_FOR_RESOURCE = "waiting_for_resource" # External resource
    WAITING_FOR_APPROVAL = "waiting_for_approval" # Human approval
    RATE_LIMITED = "rate_limited"                 # Hit rate limit
    CIRCUIT_OPEN = "circuit_open"                 # Circuit breaker open
    BLOCKED = "blocked"                           # Explicitly blocked
    CUSTOM = "custom"                             # Your own reason

Tracking Wait Time

# How long has agent been waiting?
duration = tracker.get_wait_duration("writer")
print(f"Waiting for {duration.total_seconds():.1f}s")

# Check if waiting too long
if tracker.is_waiting_too_long("writer", max_seconds=60):
    print("Writer has been waiting over a minute!")

Getting All Waiting Agents

# Who's waiting?
waiting = tracker.get_all_waiting()
# {
#   "writer": {"reason": "waiting_for_input", "source": "researcher", "since": "..."},
#   "reviewer": {"reason": "waiting_for_approval", "source": "human", "since": "..."},
# }

# Just the count
count = tracker.count_waiting()  # 2

Wait Chain Analysis

Find wait dependencies:

# Who is waiting for whom?
chain = tracker.get_wait_chain()
# {
#   "writer": ["researcher"],
#   "reviewer": ["writer", "researcher"],  # Transitive
# }

# Detect deadlocks (circular waits)
deadlock = tracker.detect_deadlock()
if deadlock:
    print(f"Deadlock detected: {deadlock}")
    # ["agent_a", "agent_b", "agent_a"]  # A→B→A

Events

def on_wait_start(agent_id, reason, source):
    print(f"{agent_id} started waiting for {source}")

def on_wait_end(agent_id, duration):
    print(f"{agent_id} stopped waiting after {duration}s")

def on_long_wait(agent_id, duration):
    print(f"Warning: {agent_id} waiting for {duration}s")

tracker.on_wait_start(on_wait_start)
tracker.on_wait_end(on_wait_end)
tracker.on_long_wait(on_long_wait, threshold_seconds=30)

Integration with Cloud

When connected to Splinter Cloud, wait tracking enables:

• Live dashboard — See waiting agents in real-time
• Bottleneck detection — Identify slow agents
• Deadlock alerts — Automatic notification

s = Splinter(openai_key="...", api_key="sk-splinter-...")
# Wait tracking data syncs to cloud automatically

Best Practices

Always track wait reasons

Makes debugging much easier.

Set wait thresholds

Alert when agents wait too long.

Monitor for deadlocks

Especially in complex workflows.

Include source information

“Waiting for X” is more useful than just “waiting”.