Module invoke_output 

CP-11 fix-3: bounded, budgeted collection of command.invoke output.

Threat model

An interactive plugin command streams command.output / task.* notifications to the host while its command.invoke JSON-RPC response is pending. The pre-fix architecture handed the producer an mpsc::UnboundedSender<InvokeCommandEvent> whose consumer (the TUI) drained it only AFTER the call resolved — so a hostile or malfunctioning extension could park an aggregate flood of output bytes in host memory for the whole invocation window. The 4 MiB per-frame reader cap bounds individual frames, the 120 s invoke timeout bounds duration, and the one-in-flight user action bounds concurrency — none of them bounds aggregate queued bytes.

Architecture

A naive bounded-channel substitution would deadlock: the invocation loop awaits sends into the sink while the TUI’s post-hoc drain waits for the call to resolve. Instead, the sink is BOUNDED (INVOKE_EVENT_QUEUE_CAPACITY) and paired with an EAGERLY CONCURRENT collector that runs alongside the invocation (see ExtensionManager::invoke_command_collected, which joins both futures). The collector consumes unconditionally — it never waits on anything but the channel — so producers are always paced, never parked behind an absent consumer, and budgets are enforced at PRODUCTION time:

• retained payload bytes are capped by InvokeOutputBudget::max_retained_payload_bytes, with the final partially-fitting text event truncated UTF-8-safely via agent_core::BoundedText;
• retained events are capped by InvokeOutputBudget::max_retained_events;
• terminal/control semantics are preserved past budget exhaustion through a small dedicated control reserve: the first CommandOutputEvent::Done marker is always retained, and a task.done for a task whose task.start was retained is retained (at most once per task id) so no retained task is left dangling;
• everything else past budget is dropped WHOLE with exact byte/event accounting; dropped Error events are additionally counted so callers can surface error visibility (see InvokeOutputReport::limit_notice).

Truncation/coalescing policy is deterministic: head retention in arrival order; only Text/System/Error content is partially truncated (single-string payloads); Table and task.* events never fit partially — they are retained whole or dropped whole.

Cancellation/timeout/drop safety: the sink is single-ownership (not Clone); when the invocation future is dropped (caller cancellation, 120 s timeout, transport failure) the sink drops with it, the collector’s channel closes and InvokeOutputCollector::collect returns promptly, and any producer blocked on an awaited send is released by the dropped receiver. No detached tasks are involved.

Counter identities (asserted by tests):

• consumed_payload_bytes == retained + truncated + dropped bytes;
• consumed_events == retained_events + dropped_events (truncated events are retained);
• produced_* >= consumed_*, equal while every send succeeds;
• retained_payload_bytes <= max_retained_payload_bytes + control_reserve_bytes at every instant.

Structs

InvokeEventSink

Bounded, metered producer handle for command.invoke events.

InvokeOutputBudget

Invocation-local budgets applied by the collector at production time.

InvokeOutputCollector

Eagerly concurrent consumer for one invocation’s event stream.

InvokeOutputCounters

Exact invocation-local accounting. See module docs for identities.

InvokeOutputNotice

Limit notice for UI surfaces.

InvokeOutputReport

Everything retained from one command.invoke output stream, plus exact accounting.

InvokeSinkClosed

Error returned by InvokeEventSink::send when the collector (and its channel) is gone. Producers treat this exactly like the old closed-unbounded-sink case: stop forwarding, let the call finish.

Constants

INVOKE_CONTROL_RESERVE_BYTES

Byte reserve usable ONLY by post-budget control events (Done markers, task.done completion for already-retained tasks) so terminal semantics survive budget exhaustion. Its consumers are structurally bounded: at most one task.done per retained task id plus one Done marker.

INVOKE_EVENT_QUEUE_CAPACITY

Bounded producer→collector handoff capacity, in events. Small on purpose: with an eagerly-draining collector, channel-parked bytes are scheduling-transient; capacity only smooths bursts (mirrors NOTIFICATION_QUEUE_CAPACITY in the transport underneath).

INVOKE_RETAINED_BYTE_BUDGET

Default invocation-local retained payload byte budget. Command output is chat-transcript text; 256 KiB comfortably covers legitimate commands while capping hostile floods.

INVOKE_RETAINED_EVENT_BUDGET

Default invocation-local retained event budget (guards against floods of many tiny events — e.g. hostile task.start storms — that stay under the byte budget).

Functions

event_payload_bytes

Payload byte weight of one event: the UTF-8 byte length of every producer-controlled string it carries. Done is a zero-byte marker.

invoke_event_channel

Create the bounded sink/collector pair for one command.invoke.