Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func IsLifecycleEvent ¶
IsLifecycleEvent reports whether an event must be flushed immediately by a coalescing sink (see Buffered). Lifecycle events are step/workflow transitions, resume lifecycle markers, drops, errors, plan-ready, transcript-sealed, and coordinator-inbox (reverse-reply) messages - any signal where user-visible latency matters more than batching. High-frequency delta events (narration, agent_turn, tool_call, agent_inbox_drain, idle/wake, coordinator_message/synthesis, generic message) are NOT lifecycle: they are safe to coalesce within the Buffered window. Exported so consumers (TUI adapter, JSONSink) share one canonical list rather than re-encoding it.
Types ¶
type BufferedSink ¶
type BufferedSink struct {
// contains filtered or unexported fields
}
BufferedSink is a ProgressSink that batches high-volume delta events (narration, agent_turn, tool_call, output, etc.) within a time window and flushes immediately on lifecycle events (see IsLifecycleEvent). The returned sink implements zenflow.ProgressSink and an additional Close error method that flushes any pending batch - call it when the run ends to avoid losing the final window of deltas. Backpressure: OnEvent/OnOutput block when the internal buffer is full (defaults to 1024 entries); consumers wrapping a slow sink should choose a larger buffer to avoid coupling LLM throughput to UI responsiveness. Goroutine model: a single dispatcher goroutine owns the batch and the timer. Callers submit work via a buffered channel; lifecycle events cause the dispatcher to emit the current batch (in order) followed immediately by the lifecycle event itself.
func (*BufferedSink) Close ¶
func (b *BufferedSink) Close() error
Close drains any pending batch and stops the dispatcher. Safe to call multiple times. Order matters: close(b.done) runs BEFORE the flush dance so concurrent OnEvent/OnOutput callers prefer the wrapped (synchronous) path instead of racing into b.in. The loop also drains any items that snuck into b.in after isFlush was queued (FIFO would otherwise leave them stuck behind the flush sentinel and silently dropped when the loop returns).
type ClosableProgressSink ¶
type ClosableProgressSink interface {
zenflow.ProgressSink
Close() error
}
ClosableProgressSink is a ProgressSink that owns shutdown semantics. Exposed so callers of Buffered can hold the concrete shutdown contract via an exported named interface instead of the anonymous interface{ Close error } type assertion ceremony. defer Close is the canonical shutdown pattern; the interface lets callers store the result without knowing the unexported *BufferedSink type.
func Buffered ¶
func Buffered(wrapped zenflow.ProgressSink, window time.Duration) ClosableProgressSink
Buffered returns a ProgressSink that coalesces delta events within the given window. When window <= 0, defaults to 100ms. The return type is ClosableProgressSink so callers can defer b.Close directly without an anonymous type assertion. The underlying *BufferedSink already exposes Close error; this just surfaces the contract through the static type.
type JSONSink ¶
type JSONSink struct {
// contains filtered or unexported fields
}
JSONSink writes workflow events as NDJSON (one JSON object per line). It is safe for concurrent use. Construct via JSON(w). The concrete type is exported so callers can inspect Err after the run completes to detect I/O failures the streaming write path silently latched.
func JSON ¶
JSON returns a ProgressSink that writes NDJSON to the given writer. It is the canonical machine-readable sink for `zenflow flow --json` and any library consumer that wants structured event output.
func (*JSONSink) Err ¶
Err returns the first write error encountered, if any. Consumers should check after workflow completion to detect I/O failures.
Source Files