Proposal: Framework-level `safe_wait` primitive for interruptible blocking tools

[Leoyzen]

Summary

Propose a framework-level safe_wait() async utility that lets any tool block on external conditions while remaining responsive to urgent (steer) messages. This replaces the current asyncio.sleep(1) polling pattern used in dynamic team mode watch tools (list_blackboard(watch=True), team_status(watch=True)), and provides a reusable primitive for any downstream tool that needs to wait.

Problem

Current State

In the feat/dynamic-team-mode worktree, team tools implement a watch parameter using a blocking polling loop:

# capabilities/team_comm_capability.py — current implementation
while time.monotonic() < deadline:
    await asyncio.sleep(1)
    current = set(team_state.list_blackboard(team_id))
    if current != initial:
        break

Issues:

1. Blocks the entire agent turn — up to 300s, agent can't do anything else
2. Urgent messages can't get through — steer/followup messages queue behind the blocking tool, must wait for watch to finish
3. Wastes resources — LLM API connection held open, CPU on 1s polling
4. Not reusable — each tool reimplements the same polling pattern

Cross-Framework Survey

Surveyed 8 frameworks (opencode, deer-flow, hermes-agent, oh-my-opencode, qwen-code, pi, zed, pydantic-ai-harness). Key findings:

• No framework has a "watch" tool primitive — all use either polling (oh-my-opencode, hermes), event-driven waiting (qwen-code waitForTeammateActivity, zed watch::channel + select!), or deferred execution (pydantic-ai-harness Monty sandbox)
• No framework interrupts mid-tool for message injection — all deliver messages at safe points (turn boundaries, tool batch boundaries). qwen-code's enqueueMessage queues during RUNNING, processes at IDLE
• qwen-code's waitForTeammateActivity() is the closest pattern — a deferred Promise that resolves on message/terminated/timeout/abort
• zed's watch::channel + futures::select! is the gold standard for multi-source interruptible waiting

Proposal

Design: "Safe Wait" — block but let urgent messages through

Core principle: tool stays blocking (turn doesn't end), but steer (urgent) can interrupt the wait; followup (non-urgent) queues naturally for the next turn.

This maps to the existing steer/followup dual delivery model:

• steer() = urgent, mid-turn injection → interrupts wait
• followup() = non-urgent, next-turn queue → does NOT interrupt, queues naturally
• Tool's watch conditions (file changes, events, etc.) → wait succeeds
• Timeout → wait expires

Public API

# agentpool/agents/safe_wait.py

from agentpool.agents.safe_wait import safe_wait, SafeWaitResult

@dataclass(frozen=True)
class SafeWaitResult:
    reason: Literal["condition", "steer", "timeout"]
    value: Any = None
    steer_content: str | None = None

async def safe_wait(
    ctx: RunContext[AgentContext],
    timeout: float,
    conditions: list[Awaitable[Any]] | None = None,
) -> SafeWaitResult:
    """Block until: any condition fires, steer interrupts, or timeout.
    
    Non-urgent (followup) messages queue naturally for the next turn.
    """
    ...

Usage (downstream tools)

from agentpool.agents.safe_wait import safe_wait

async def list_blackboard(self, ctx, watch=False, timeout=300):
    if not watch:
        return immediate_result
    
    # Build condition: poll file changes
    file_changed = asyncio.Event()
    initial = self._snapshot_files(team_id)
    
    async def poll():
        while not file_changed.is_set():
            await asyncio.sleep(2.0)
            if self._snapshot_files(team_id) != initial:
                file_changed.set()
    
    poll_task = asyncio.create_task(poll())
    try:
        result = await safe_wait(ctx, timeout=timeout, conditions=[file_changed.wait()])
        
        keys = team_state.list_blackboard(team_id)
        match result.reason:
            case "condition":
                return ToolReturn(return_value=f"<watch: changes detected>\n{keys}")
            case "steer":
                return ToolReturn(return_value=f"<watch interrupted by urgent: {result.steer_content}>\n{keys}")
            case "timeout":
                return ToolReturn(return_value=f"<watch timeout ({timeout}s)>\n{keys}")
    finally:
        poll_task.cancel()

Message Flow

Lead calls list_blackboard(watch=True, timeout=120)
  → tool calls safe_wait(ctx, timeout=120, conditions=[file_changed.wait()])
  → asyncio.wait(interrupt_queue, file_change_task, timeout=120)
  ↓ (waiting, turn stays active, zero CPU)

  ┌─ User sends urgent message → steer() → queue.put(("steer", "stop!"))
  │   → safe_wait returns reason="steer" → tool returns "interrupted by urgent: stop!"
  │   → lead processes message in same turn
  │
  ├─ Member sends non-urgent message → send_message(mode=QUEUE) → followup()
  │   → does NOT interrupt safe_wait → message queues for next turn
  │   → after watch ends (timeout/change), next turn delivers the message
  │
  ├─ Member calls write_blackboard() → file changes
  │   → 2s later file_changed event fires → safe_wait returns reason="condition"
  │   → tool returns "changes detected" + new keys
  │   → lead sees changes in same turn
  │
  └─ 120s timeout
      → safe_wait returns reason="timeout" → tool returns "timeout" + current keys

Implementation

Files to change (3 framework files + 1 new)

File	Change	Visibility
agents/safe_wait.py	NEW — safe_wait() function + SafeWaitResult dataclass	Public API
sessions/models.py	+1 field watch_interrupt: asyncio.Queue | None = None	Internal
orchestrator/run.py	steer() +3 lines to put to watch_interrupt queue	Internal
agentpool/__init__.py	Export safe_wait, SafeWaitResult	Public API

Internal mechanism

1. safe_wait() sets session.watch_interrupt = asyncio.Queue() on entry
2. RunHandle.steer() checks session.watch_interrupt and puts ("steer", content[:200]) if set
3. RunHandle.followup() is not modified — non-urgent messages queue naturally
4. safe_wait() races interrupt_queue.get() + user conditions + timeout via asyncio.wait(FIRST_COMPLETED)
5. On exit (finally): session.watch_interrupt = None, cancel pending tasks

Standalone mode fallback

When no session_pool is available (standalone agent.run()), safe_wait degrades to pure condition racing without steer support — just asyncio.wait(conditions, timeout=timeout).

Design Decisions

Why not interrupt for ALL messages?

Interrupting on every message (including routine member updates) fragments the lead's thinking. The lead would have to: process message → decide next step → re-call watch → repeat. Only urgent (steer) messages should interrupt; non-urgent ones queue naturally.

This aligns with all surveyed frameworks — none interrupt mid-tool for non-urgent messages. qwen-code queues via enqueueMessage (RUNNING → queue, IDLE → process). pi checks steering between tool batches, not mid-tool.

Why not use deferred tools (checkpoint/resume)?

Deferred tools (deferred=True, strategy="block") end the turn and checkpoint the session. This is more robust (crash recovery) but:

1. Tool function never executes (deferred bridge intercepts before execution) — can't register watch conditions
2. Requires changes to deferred_bridge.py, CheckpointManager, resume flow — high infrastructure cost
3. Overkill for the core problem (blocking prevents message delivery)

Deferred tools remain a valid future enhancement for crash-safe watch, but safe_wait solves the immediate problem with minimal changes.

Why not use EventBus?

EventBus has replay buffers that would deliver historical events on subscribe, causing false triggers. The watch_interrupt queue on SessionState is simpler, has no replay, and is scoped to the exact session.

Alternatives Considered

Approach	Pros	Cons	Verdict
safe_wait (this proposal)	Simple, reusable, 3-file change, urgent messages get through	No crash recovery, 2s poll latency for file changes	✅ Recommended
Deferred tool (checkpoint)	Crash recovery, clean session parking	Tool function doesn't run, high infra cost, overkill	Future enhancement
Non-blocking + send_message	Simplest, no blocking at all	Lead must end turn, loses "wait in same turn" semantics	Alternative for non-blocking tools
EventBus subscription	Event-driven, no polling	Replay buffer issues, deprecated descendants scope	Rejected
watch_interrupt for ALL messages	Zero latency for all messages	Fragments lead's thinking, churn	Rejected

Scope

• In scope: safe_wait() primitive, SessionState.watch_interrupt field, RunHandle.steer() hook
• Out of scope: Deferred tool integration, crash recovery, EventBus-based triggers, watchdog filesystem events (current 2s polling is sufficient for team collaboration)

Related

• RFC-0055: Dynamic Team Mode (docs/rfcs/draft/RFC-0055-dynamic-team-mode.md) — the watch parameter on list_blackboard/team_status that this proposal replaces
• Existing deferred tool infrastructure: tools/base.py (deferred=True), deferred_bridge.py, CheckpointManager — future integration target
• RunHandle.steer() / followup() — the dual delivery model this proposal hooks into