Signals

The signal channel is the runtime’s interrupt primitive: a per-session mailbox of named FIFO queues. Code inside a run parks on runtime.wait_for_signal(session_id, name); code outside the run — a web handler, a Slack callback, an operator console — unblocks it with agent.signal(session_id, name, payload). The payload crosses over and the run continues.

from loomflow import Agent
from loomflow.runtime import SqliteRuntime
 
runtime = SqliteRuntime("./journal.db")
agent = Agent("...", model="claude-opus-4-7", runtime=runtime)
 
# Somewhere inside the run (an approval handler, a custom
# architecture) the task parks:
payload = await runtime.wait_for_signal("sess-42", "approval")
 
# ...until your application delivers, from anywhere in the process:
await agent.signal("sess-42", "approval", {"action": "allow"})

Both InProcRuntime and JournaledRuntime (and its SqliteRuntime / PostgresRuntime subclasses) implement the channel.

The three entry points

# Deliver — enqueue payload under name, wake one waiter.
await agent.signal(session_id, name, payload=None)      # convenience
await runtime.signal(session_id, name, payload)         # direct
 
# Park — return the oldest payload for name, waiting until one arrives.
payload = await runtime.wait_for_signal(session_id, name)
 
# Poll — pop a queued payload if present; never blocks.
payload = runtime.poll_signal(session_id, name)          # sync, -> Any | None

agent.signal(...) is a delegate to runtime.signal(...) so callers don’t reach into runtime internals. On a runtime without signal support it logs a warning and drops the signal (no-op, no exception).

Delivery semantics

• FIFO per name. Multiple payloads under one name queue in order; wait_for_signal / poll_signal pop oldest-first.
• One waiter per payload. Multiple tasks parked on the same name are woken oldest-first, one per delivered payload. Which waiter gets which payload is scheduler-dependent, but no payload is ever lost or delivered twice.
• Queue-before-wait works. A signal delivered before anyone waits is queued; the next wait_for_signal returns immediately.
• Names and sessions are isolated. A signal for ("a", "sig") never surfaces on session "b" or under name "other".
• Cancellation-safe. A waiter cancelled after being woken but before consuming hands its wakeup to the next parked waiter, so the payload is never stranded.

Mailbox lifecycle

The mailbox lives inside the runtime’s session state:

1. Created lazily — the first signal(...) or wait_for_signal(...) for a session_id creates its mailbox, even before the session context is formally opened. Early signals queue.
2. Shared across nested contexts — re-entering runtime.session(session_id) (as concurrent runs on one session do) reference-counts the same mailbox.
3. Discarded on last exit — when the session’s last open context exits, the mailbox (and any undelivered signals) is dropped, so the sessions dict can’t grow without bound across runs.

The park-and-resume pattern

The main consumer is human-in-the-loop approval. Approval handlers are async and the signal channel is public, so a handler can park the run until an out-of-band decision arrives — no framework support needed beyond the signal API:

from loomflow.architecture import ApprovalDecision
 
async def approve(call, user_id):
    await notify_slack(call, user_id)              # out-of-band ask
    payload = await runtime.wait_for_signal(       # park the run
        session_id, "approval"
    )
    return ApprovalDecision(action=payload["action"])
 
# The run is now parked mid-turn. Your Slack webhook unblocks it:
await agent.signal(session_id, "approval", {"action": "allow"})

The model’s pending tool call waits, the transcript stays intact, and the loop continues the moment the decision lands. See Approval handlers for the full ApprovalDecision surface (edit-args, remember-decision).

Inside a custom architecture

Architectures park via the wait_for_user_signal helper, which also emits an interrupt.waiting architecture event so streaming UIs can show a “waiting for input” state:

from loomflow.architecture import wait_for_user_signal
 
payload = await wait_for_user_signal(
    deps, session.id, "resume", emit=emit,
)

When the runtime has no signal support, the helper logs a warning and returns None immediately — the run proceeds rather than deadlocking on a channel that can never be written.

Waiting with a timeout

wait_for_signal parks indefinitely. Bound it with your async library’s timeout primitive:

import anyio
 
try:
    with anyio.fail_after(300):                     # 5-minute approval SLA
        payload = await runtime.wait_for_signal(sid, "approval")
except TimeoutError:
    payload = {"action": "deny"}                    # fail closed

For long approval windows that outlive the process, don’t park at all: persist the pending request, end the run, and start a new one when the decision arrives — combine checkpoints with agent.resume(...) to continue the transcript where it stopped.