ADR-0017: Worker Claim-Loop Stays Sync; Streaming via Daemon Threads

Status: Accepted
Date: 2026-07-06

Context

M3 needed three things from the agent subprocess:

Streaming stdout — every line emitted by the agent must land as an AgentMessage graph node in real-time (so users can runex agent messages <run_key> --follow and see progress).
Cancellation — runex work cancel <work_key> must kill the subprocess within ~1 second (so users don’t wait 5 minutes for a misbehaving agent to finish).
Backwards compatibility — every existing dispatch handler (test_agent_dispatch_subprocess.py has 6 of them; Tier-2 extensions like test_extension_discovery.py add more) must keep working with no signature change.

The previous implementation was subprocess.run(...) blocking — none of these were possible.

Two paths forward

The natural OpenOPC-style answer: rewrite the worker claim-loop as async (asyncio.create_subprocess_exec, async handlers, async SQLite). This gives full streaming, full cancellation, full control.

The conservative answer: keep the worker claim-loop sync; do the streaming + cancellation inside the provider via Popen + daemon threads. The sync loop keeps calling provider.run(envelope), which now internally spawns Popen + threads + polls.

We chose the conservative path. This ADR records why and the constraints we accept.

Decision

Worker claim-loop stays synchronous (WorkerLoop._process → sync handler() → sync provider.run()). All streaming and cancellation is implemented inside SubprocessAgentProvider.run() via subprocess.Popen + 2-3 daemon threads + a queue + a 0.2s main-thread poll.

Alternatives

方案 A — Full async rewrite (rejected)

WorkerLoop becomes asyncio.Task; handler becomes async def; Store.isolated() per thread (ADR-0010 already provides this); the provider returns an async iterator over stdout lines.

Pros:

Full streaming, full cancellation, full async composition
Matches OpenOPC’s architecture exactly
Can fan-out concurrent sub-tasks naturally

Cons (the reason rejected for v1):

Breaking change for every Tier-2 provider. test_extension_discovery.py registers a custom provider whose run(self, envelope) is sync. Async rewrite means every Tier-2 author has to migrate.
Breaking change for the SQLite story. Store.isolated() exists per AGENTS.md §10 (“Thread-local Store 隔离”) but is rarely used; forcing async means every set_field call needs isolated(), otherwise thread-of-async-task gets the wrong connection.
ReactiveBus impact. ADR-0014’s effect_scope() cascades are sync. Async-rewriting the worker means they need an async variant too, otherwise the bus fires into a sync context from an async one — easy to deadlock.
Debugging. pdb works in sync code. Async debugging is much harder. Most runex users are not async-native.
Two-week tax for the rewrite vs one-week for the daemon-thread approach. ADR-0012’s “Agent dispatch via DSL primitive” already documented the conservative stance; this follows the same line.

方案 B — Mixed: keep WorkerLoop sync, add Popen + daemon threads inside provider (chosen, ADR-0017)

SubprocessAgentProvider.run() is rewritten from subprocess.run(...) to subprocess.Popen(...) + two daemon threads (_drain(stdout), _drain(stderr)) that push lines onto a queue.Queue. The main thread (still sync) polls:

Drain any pending lines from the queue → emit AgentMessage (uses the same sqlite connection as the rest of the dispatch — serialised through the queue so all writes happen on the loop thread, not the drain threads)
Check AgentRun.cancelled field (one indexed SELECT) — if True, proc.kill(), break loop
Check timeout — if elapsed > self.timeout, proc.kill(), break loop
time.sleep(0.2)

After proc exits: drain remaining queue (timeout 2s), join threads.

Pros:

Zero changes to the worker claim-loop — WorkerLoop, _process, all Tier-2 handlers keep working
Streaming works — every line lands as AgentMessage within one poll interval (≤ 0.2s) of being written
Cancellation works — proc.kill() runs within one poll interval
SQLite connection is owned by the main thread — no Store.isolated() needed; the drain threads never touch SQLite directly, they put items on the queue
Backwards compat — Provider.run(envelope) keeps its sync signature. Tier-2 providers don’t change.

Cons:

Per-thread idle CPU — time.sleep(0.2) means up to 5 wakeups per second per running agent. With 10 concurrent agents, 50 wakeups per second total. Negligible.
Cancel latency = 0.2s — slightly worse than OpenOPC’s ~immediate SIGINT. Acceptable for v1; documented as a follow-up tightening target.
Drain threads don’t see SQLite writes from the main thread in real-time. They push to the queue; the main thread reads the queue and writes SQLite. So the agent sees its own messages as “queued” until the main thread processes them. 0.2s max delay.
openclaw cancel doesn’t reach Gateway (covered in docs/limitations.md §7.2) — openclaw’s design requires a chat.abort RPC after proc.kill(), which the daemon-thread approach does NOT do. Documented separately.

方案 C — `asyncio` for the provider only, keep sync worker (considered, deferred)

The provider itself uses asyncio.create_subprocess_exec() + async queue; the worker loop calls provider.run() via asyncio.run() in a thread-pool executor. The provider’s API stays sync.

This is the best of both worlds IF asyncio works cleanly with our SQLite story. It doesn’t (yet) — Store.isolated() exists but isn’t default; the reactive bus is sync; etc. Deferring to v2.

Consequences

What becomes easier

Tier-2 extensions work without changes. A provider author writing def run(self, envelope) -> AgentResult: doesn’t need to know about streaming or cancellation. The default subprocess provider handles both; they can opt out by providing their own.
Adding new dispatch handlers (e.g. a future remote executor) doesn’t have to know about subprocess plumbing — just return AgentResult.
M3 implementation footprint: ~280 lines in subprocess_agent.py, ~150 in test_subprocess_popen.py. If we’d gone async, the footprint would be ~800 lines across subprocess_agent.py, worker/loop.py, reactive.py, plus schema and dispatch handler migrations.

What becomes harder

Cancel latency is bounded by poll interval (default 0.2s). For most users this is invisible. For tight loops (sub-second kills), we’d need to drop the poll interval or move to inotify-style file events on a side channel. Defer.
openclaw Gateway-mode cancel is incomplete. Documented in docs/limitations.md §7.2 — needs a follow-up that calls openclaw chat.abort --run-id <id> after proc.kill(). The daemon-thread plumbing is the right shape for this; we just haven’t wired the protocol.
One process per agent means file-descriptor pressure. Each Popen consumes ~3 fds (stdin/stdout/stderr). 100 concurrent agents = 300 fds. Linux default is 1024 per process. Fine for now; documented as “watch the limit if you scale to 500+ concurrent”.
The sync worker can’t preempt an in-flight provider.run() that takes > poll_interval to start. If a Tier-2 provider blocks for 10 seconds inside its sync run() (e.g. a network call), the worker can’t even check cancel during that window. Out of scope for the subprocess provider (which polls internally), but a real risk for custom providers. Documented in the Provider ABC contract.

When we would revisit

Documented trigger conditions — if any of these becomes true, this ADR is reopened:

Average concurrent agent count > 50. The 0.2s × N polling starts to show CPU. Move to inotify or async.
Cancel latency requirement drops below 100ms. Today 200ms max. Need inotify or signalfd.
Agent providers need to compose async (e.g. stream from a websocket, fan-out concurrent sub-tasks). Sync interface becomes a bottleneck.
Store.isolated() becomes default (i.e. the architectural stance shifts to “every thread is its own connection”). At that point async rewrite is much cheaper.