Signals

Signals let a running agent pause and wait for input from outside the run, such as a human clicking “approve” in Slack, and then carry on with that input. Use them for human-in-the-loop steps where the run should stay alive while it waits.

The signal channel is 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
# (TimeoutError after timeout_s seconds; omit it to wait indefinitely).
payload = await runtime.wait_for_signal(session_id, name, timeout_s=300)
 
# 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. Early signals wait, within bounds. A signal(...) for a session that isn’t open (sent before its run starts, or after it ended) is held in a pending mailbox, and the run picks it up when it opens the session. Pending signals are bounded: they’re kept for at most max_pending_signal_sessions sessions (default 10,000; the least recently used session’s signals are dropped first) and for pending_signal_ttl_s seconds (default 3,600). Signals sent to sessions that never run can’t grow memory without limit.
2. Shared across nested contexts. Re-entering runtime.session(session_id) (as nested or 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. A wait_for_signal(...) on a session that isn’t open holds the mailbox while it waits; when it returns or times out, signals still queued go back to the pending mailbox.

Both bounds are constructor arguments of InProcRuntime and JournaledRuntime; pass None to lift one. SqliteRuntime and PostgresRuntime use the defaults, so wrap a store yourself to change them:

from loomflow.runtime import InProcRuntime, JournaledRuntime, SqliteJournalStore
 
runtime = InProcRuntime(max_pending_signal_sessions=1_000, pending_signal_ttl_s=600)
# or, durable:
runtime = JournaledRuntime(
    store=SqliteJournalStore("./journal.db"),
    max_pending_signal_sessions=1_000,
    pending_signal_ttl_s=600,
)

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, with 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
    try:
        payload = await runtime.wait_for_signal(   # park the run
            session_id, "approval", timeout_s=300,
        )
    except TimeoutError:
        return ApprovalDecision(action="deny")     # fail closed
    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, so the run proceeds rather than deadlocking on a channel that can never be written.

Waiting with a timeout

Without timeout_s, wait_for_signal parks until a signal arrives. Pass timeout_s= to bound the wait; it raises TimeoutError when the time is up:

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

wait_for_user_signal takes no timeout; wrap it in anyio.fail_after if an architecture needs one.

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.