Documentation
¶
Overview ¶
Package streaming provides the partial-frame sink used by tools that opt into streaming dispatch (roadmap P3-28 first half).
The motivating problem: long-running tools (`subghz_receive`, `wifi_wardrive_start`, `subghz_rx_raw`) run for tens of seconds or minutes and currently block dispatch until they finish. Operators get no live feedback; the agent can't surface intermediate findings ("got a handshake — operator may want to stop early"); a wedged tool is indistinguishable from one making slow progress.
The Sink type is the contract a streaming tool emits frames against. Each Frame is one quantum of partial output — a parsed scan-line, a periodic progress tick, a structured event. The host (agent / MCP server / test harness) attaches a consumer that forwards frames to operator-facing channels (CLI status line, web UI, SSE stream) without affecting the LLM-facing tool_result, which is the tool's final return value.
Design notes:
- Channel-backed. A bounded buffer drops frames on overflow rather than blocking the tool — operator-facing live feedback is best-effort, not load-bearing.
- Frames carry a sequence number so consumers can detect drops.
- Close is idempotent. Tools defer Close on the sink; consumers range over Frames() and exit when the channel closes.
- Nil-safe sentinel. A nil *Sink's Send is a no-op so streaming handlers can run unconditionally — falling back to a normal non-streaming dispatch is just "pass nil here".
Index ¶
Constants ¶
const DefaultBufferSize = 256
DefaultBufferSize is the channel capacity used by NewSink when the caller passes 0. Sized so a chatty parser (one frame per scanned AP, ~100 APs in a 30-second window) can run without dropping anything against a slow consumer that drains every 0.5s.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Frame ¶
Frame is one quantum of partial tool output. Tool is the canonical tool name; Seq is a 1-based monotonically-increasing per-sink counter so consumers can detect drops. Bytes is the payload (caller-defined shape — JSON for structured tools, plain text for progress lines). Time is the moment the tool emitted the frame in UTC.
type Handler ¶
Handler is the streaming-handler signature: a function that pumps frames at a Sink while computing a final string return value (the LLM-facing tool_result). Mirrors tools.Handler with one extra parameter so opt-in tools don't need to break the non-streaming contract.
The handler MUST defer sink.Close() so consumers can exit cleanly even on early returns / panics. The host wires the consumer (operator-facing live UI, SSE forwarder, audit observer) before calling the handler.
type Sink ¶
type Sink struct {
// contains filtered or unexported fields
}
Sink is the per-call frame channel a streaming tool writes to. Construct with NewSink. A nil *Sink is a usable "no-op" sentinel — every method short-circuits — so dispatch code can pass nil for non-streaming tools without an "if sink != nil" wrapper at every emission point.
func NewSink ¶
NewSink constructs a Sink for the given tool name. buffer ≤0 falls back to DefaultBufferSize. The returned Sink is ready for Send calls immediately; callers should defer Close.
func (*Sink) Abort ¶ added in v0.56.0
func (s *Sink) Abort()
Abort signals the producer to wrap up early. The channel returned by Aborted() is closed; idempotent and nil-safe. A producer honouring abort polls Aborted() in its select loop (or relies on the per-call context that dispatch cancels in tandem) and returns promptly with whatever partial result it can summarise. Send is NOT short-circuited — producers may still emit a final summary frame between observing Aborted() and calling Close.
func (*Sink) Aborted ¶ added in v0.56.0
func (s *Sink) Aborted() <-chan struct{}
Aborted returns a channel that is closed when Abort has been called. Producers should select on this alongside ctx.Done() so abort fires regardless of whether dispatch cancels the context. A nil *Sink returns a nil channel — selecting on a nil channel never receives, so non-streaming dispatch (sink=nil) naturally has no abort signal.
func (*Sink) Close ¶
func (s *Sink) Close()
Close stops accepting new frames and closes the underlying channel so consumers `range`-looping over Frames() exit. Idempotent — safe to defer.
sendMu is held during the closed-flag store and the channel close so a concurrent Send is guaranteed to either complete before the close (frame delivered) or observe closed=true after the lock hands off (frame rejected). Without this pairing a Send that passed s.closed.Load()==false could race the close and panic.
func (*Sink) Drops ¶
Drops reports how many Send calls were rejected because the buffer was full. Zero on a healthy stream; non-zero indicates the consumer is slower than the producer and the operator may want to attach a faster sink.
func (*Sink) Frames ¶
Frames returns the receive-only channel consumers range over. Closing happens via Sink.Close. A nil *Sink returns a nil channel (a `range` over nil blocks forever — callers should guard with a nil-sink check before installing a consumer).
func (*Sink) IsAborted ¶ added in v0.56.0
IsAborted is a non-blocking convenience for producers that prefer a periodic poll over a select. nil-safe; returns false on a nil sink.
func (*Sink) Send ¶
Send pushes a frame onto the sink. Returns false when the sink is nil, already closed, or its buffer is full (in which case the frame is dropped and the internal Drops counter advances). Send never blocks — operator-facing live feedback must NOT slow the tool's actual work.
Concurrency: safe for use from multiple goroutines on the same sink, but the typical pattern is one producer goroutine per Sink. sendMu serialises against Close so a Send racing past the s.closed.Load() check cannot panic on a just-closed channel.
Source Files