fix(sdk/python): stop the structured log stdout write from blocking the event loop - #1066
Merged
Merged
Conversation
…ines A single daemon thread drains a bounded FIFO of (stream, line) pairs. Deferral only engages where it can help — the caller is on a running event loop and the destination has a real file descriptor — so synchronous callers and in-memory captures keep writing inline and stay immediately visible. Under back-pressure the queue discards its oldest pending lines instead of blocking the producer, and the writer emits one log.dropped record per destination naming the count, so loss is never silent. An atexit drain bounded by AGENTFIELD_LOG_QUEUE_FLUSH_SECONDS covers normal shutdown, and an at-fork hook gives the child fresh state. Refs #985 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…nt loop _emit_structured_record printed inline on the calling thread, which in an agent node is the event loop, into the _TeeTextIO wrapper over a real pipe. A consumer that stops draining that pipe froze the whole loop for the duration — heartbeats, in-flight reasoners and the control-plane client alike. Size bounding did not help; the stall is the pipe, not the payload. Both stdout paths now go through the bounded writer, which also keeps their relative order. _DynamicStdoutHandler additionally drops its logging handler lock: Handler.handle() holds that lock across emit(), so a synchronous thread blocked inline on a stalled stdout would otherwise still stall an event-loop caller before it reached the queue. The writer serializes instead. Refs #985 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
_TeeTextIO holds its write lock across the blocking write to the original stream. A fork while another thread holds it leaves the child with a mutex that is locked and has no owner, so every later write in the child deadlocks and its stdout is gone for good. An after_in_child hook now gives each installed tee a fresh lock, clears the partial line inherited mid-write, re-creates the ring and follower locks and drops the parent's follower queues. There is deliberately no before= hook: acquiring those locks ahead of the fork would make fork() itself wait on a stalled pipe. The hazard predates this branch, but a dedicated writer thread that can hold the lock for the length of a consumer stall makes it easy to hit. Refs #985 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
AGENTFIELD_LOG_QUEUE, AGENTFIELD_LOG_QUEUE_SIZE and AGENTFIELD_LOG_QUEUE_FLUSH_SECONDS, alongside the existing logging variables: when deferral engages, the drop-oldest policy and its log.dropped marker, that os._exit bypasses the exit flush, and that ordering against a caller's own print() is no longer guaranteed. Refs #985 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Contributor
Performance
✓ No regressions detected |
Contributor
📊 Coverage gateThresholds from
✅ Gate passedNo surface regressed past the allowed threshold and the aggregate stayed above the floor. |
Contributor
📐 Patch coverage gateThreshold: 80% on lines this PR touches vs
✅ Patch gate passedEvery surface whose lines were touched by this PR has patch coverage at or above the threshold. |
Setting logging.Handler.lock to None relied on handle() calling self.acquire(), which skips a falsy lock. Python 3.13 changed handle() to `with self.lock:`, so None raises TypeError: 'NoneType' object does not support the context manager protocol — every plain log line on 3.13 died in the handler, and a test whose synchronous writer thread was killed that way hung CI rather than failing. Use a no-op lock object instead: it satisfies both the acquire/release and the context-manager protocols, and keeps the property we want, which is that nothing serializes on the handler while the bounded writer already does. Refs #985 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
These helpers block on a pipe read or a lock on purpose. As non-daemon threads, a failing assertion left them alive and the interpreter waited for them at exit, so a test failure presented as a hung CI job instead of a failure. As daemons the session reports the failure and exits. Refs #985 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
19 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
_emit_structured_recordand the plain-log handler both write to stdout inline, on whateverthread called them. In an agent node that thread is the event loop, and
sys.stdoutis the_TeeTextIOwrapper, so one emit takes the tee's lock, writes, appends to the process-logring, and then flushes — the real OS write. If whoever is reading the other end of the pipe
stops draining it (a busy tty, a piped consumer, a
kubectl logsreader that stalls) thewrite blocks in the kernel and the whole loop stops with it. Size bounding (#1002/#1028) does
not help: this is about the pipe, not the payload.
This adds
agentfield/log_writer.py— a bounded FIFO with one daemon writer thread that allSDK logger stdout writes go through. Deferral only engages where it can actually help: the
emit happens on a running event loop and the destination has a real file descriptor. Off the
loop, or into an in-memory capture, the write stays inline and immediately visible exactly as
before. Under back-pressure the queue discards its oldest pending lines rather than blocking
the producer, and the writer emits one
log.droppedrecord naming the count, so loss is neversilent.
AGENTFIELD_LOG_QUEUE=falserestores today's behaviour.Two supporting changes fall out of it:
_DynamicStdoutHandlerno longer creates alogginghandler lock.Handler.handle()holdsthat lock across
emit(), so a synchronous thread blocked inline on a stalled stdout wouldotherwise still stall an event-loop caller before it ever reached the queue. The writer
provides the serialization now.
node_logsregisters anos.register_at_fork(after_in_child=...)hook that gives theinstalled tees, the ring and the follower registry fresh locks in a forked child. A tee's
write lock is held across the blocking write, so a fork at the wrong moment leaves the child
with a locked mutex and no owner and deadlocks its stdout forever. That hazard predates this
PR, but a dedicated writer thread holding the lock for the length of a stall makes it easy
to hit, so it is fixed here.
@AbirAbbas — this reverses the "not doing a FIFO writer" call in
#985 (comment), so here are the
numbers you asked for, and what the design does about each of the four objections:
stream.fileno()to succeed.Under pytest capture it raises
io.UnsupportedOperation, so every captured write staysinline. The existing
capsys-based logger tests pass unchanged.os._exit— anatexitdrain bounded byAGENTFIELD_LOG_QUEUE_FLUSH_SECONDS(2 s) covers normal exit;os._exitstill bypasses itand the docs say so, with
AGENTFIELD_LOG_QUEUE=falseas the answer for a process that needsinline writes before such an exit.
writer, and the backlog check keeps a synchronous caller behind anything already queued, so
their relative order holds. Ordering against a caller's own bare
print()does not, and thatis documented.
Still inline, and worth knowing: handlers this SDK does not own (uvicorn's access log, a
caller's own
print) write to stdout themselves and can still block on a stalled pipe. Thischange owns the SDK logger's writes.
Validation
A real control plane (
af server, local mode, isolatedHOME) and a real Python agent nodestarted with its stdout on a pipe. A collector thread drains that pipe, then stops draining for
N seconds while the driver calls a reasoner that logs 400 structured records from the event
loop and, concurrently, calls a second silent reasoner on the same loop.
On
mainthe reasoner's latency tracks the stall 1:1 and the second reasoner is stuck behindit. With the queue, latency is flat no matter how long the consumer is away, and all 800
mirrored lines still arrive once it comes back.
Overflow, same setup: 3000 records against a 12 s stall overflows the 1024-line queue, the
reasoner still returns in 8.4 s (7.6 s with the collector healthy), and stdout carries exactly
one marker —
{"level":"warning","event_type":"log.dropped","message":"dropped 1918 structured log lines (stdout back-pressure)","attributes":{"dropped":1918}}— ahead of the survivinglines.
A narrower in-process probe (400 records, 6 s stall, a 10 ms
asyncio.sleepwatchdog) showsthe same thing at loop level: worst watchdog lag 5369 ms on
main, low single-digit ms here.Gates run locally on the versions CI pins:
ruff check .,mypy --config-file mypy.ini agentfield/, and the full./scripts/run_pytest.shsuite.Fixes #985
🤖 Generated with Claude Code