Documentation
¶
Overview ¶
Package audio provides real-time audio monitoring for Arena duplex runs.
Observer model — read this before changing anything ¶
Audio in this package follows a strict observer model: this package exposes a fan-out bus (AudioRouter) whose subscribers (LocalSink, RMS level meters, SSE web playback, recording sinks) are read-only observers. They watch the audio flowing through the data path; they MUST NOT influence it.
Cadence (the wall-clock rate at which audio chunks advance through the pipeline) is owned by exactly one place — the runtime/pipeline/stage.AudioPacingStage on the data path itself. Everything in this package is downstream of that cadence authority and is forbidden from feeding rate signals back to it. Specifically:
- Router.Publish drops on full intake. It does not block.
- Router fan-out drops per-consumer on full inbox. It does not block.
- LocalSink.Push drops on full sink buffer. It does not block.
Drop-on-full is load-bearing, not a wart. If you change any of these to block, you create a backchannel from a UI/observer surface into the producer pipeline — and via the input branch's broadcast, into the LLM provider's session. The provider's VAD timestamps arrival cadence to time turn-end silence, so warping that cadence based on "is anyone listening locally" produces wrong turn boundaries. Two other reasons block-on-full doesn't work even before that:
- Many parallel runs typically execute headlessly with no LocalSink attached at all. Sink-driven backpressure can't be the model when most runs have no sink to be back-pressured by.
- Even when a sink is attached, the operator may switch which run they're listening to — a run's correctness can't depend on whether someone happens to be tuned in.
If you need cadence enforcement somewhere, the answer is another AudioPacingStage instance on the data path, not "make the sink blocking."
Lifecycle ¶
Per-run. Each scenario run gets its own AudioRouter goroutine, started when MonitorTap is constructed and stopped at run completion.
Flow ¶
Audio chunks flow from the duplex provider stage through MonitorTap into the router, which fans out to per-consumer bounded channels. Slow consumers drop their own frames; the router and other consumers keep flowing.
Index ¶
- Constants
- Variables
- func IsValidRate(rate int) bool
- type AudioRouter
- func (r *AudioRouter) Close()
- func (r *AudioRouter) Done() <-chan struct{}
- func (r *AudioRouter) DropCount(id string) int64
- func (r *AudioRouter) IntakeDropCount() int64
- func (r *AudioRouter) Publish(frame Frame)
- func (r *AudioRouter) RegisterDrainHandler(h DrainHandler)
- func (r *AudioRouter) Subscribe(id string, bufSize int) <-chan Frame
- func (r *AudioRouter) Unsubscribe(id string)
- func (r *AudioRouter) WaitOutputDrained(ctx context.Context)
- type Direction
- type DrainHandler
- type Frame
- type LocalSink
- type LocalSinkConfig
- type Monitor
- type MonitorConfig
- type MonitorMode
- type MonitorTap
- type MonitorTapConfig
- type Options
- type RMSFrame
Constants ¶
const ( // Rate16k is the 16 kHz canonical rate. Rate16k = 16000 // Rate24k is the 24 kHz canonical rate (default). Rate24k = 24000 // Rate48k is the 48 kHz canonical rate. Rate48k = 48000 )
Canonical sample rates supported by the router.
Variables ¶
var ValidRates = []int{Rate16k, Rate24k, Rate48k}
ValidRates lists the canonical rates accepted by the router.
Functions ¶
func IsValidRate ¶
IsValidRate returns whether the given rate is a supported canonical rate.
Types ¶
type AudioRouter ¶
type AudioRouter struct {
// contains filtered or unexported fields
}
AudioRouter is a per-run goroutine that fans audio frames out to per-consumer bounded channels. Slow consumers drop their own frames; other consumers and the router keep flowing.
All publish paths are non-blocking by design — see the package-level observer-model doc in types.go. Subscribers are observers; they may drop frames they can't keep up with, but they cannot push back on the producer. Cadence enforcement, if needed, belongs upstream in an AudioPacingStage on the data path, not here.
Lifecycle: NewAudioRouter starts the dispatch goroutine. Close stops it and closes all consumer channels. Close is idempotent. Publish is safe to call after Close — it falls through to a no-op rather than panicking on send-to-closed-channel.
func NewAudioRouter ¶
func NewAudioRouter(canonicalRate int) *AudioRouter
NewAudioRouter constructs a router and starts its dispatch goroutine. canonicalRate is the rate frames are expected to arrive at; resampling happens upstream in MonitorTap before Publish is called.
func (*AudioRouter) Close ¶
func (r *AudioRouter) Close()
Close stops the router goroutine and closes all consumer channels. Safe to call multiple times.
Order of operations is important for the no-panic guarantee on Publish: take the close write lock, set r.closed, only then close r.in. Once we hold the write lock, no Publish is in flight (each Publish takes the read lock for its full duration). After this returns, all subsequent Publish calls observe r.closed and skip the send.
func (*AudioRouter) Done ¶
func (r *AudioRouter) Done() <-chan struct{}
Done returns a channel that is closed when the router has shut down. Useful for downstream owners (e.g. audio.Monitor) to drop a router from their registry without polling, since per-run routers outlive the run only briefly.
func (*AudioRouter) DropCount ¶
func (r *AudioRouter) DropCount(id string) int64
DropCount returns the number of frames dropped for a given consumer.
func (*AudioRouter) IntakeDropCount ¶
func (r *AudioRouter) IntakeDropCount() int64
IntakeDropCount returns the cumulative number of frames dropped at the router's intake (i.e. r.in was full when Publish ran). Distinct from DropCount, which measures per-consumer drop. A non-zero value here means producers are bursting faster than the dispatch goroutine can fan out — observable, recoverable, not a hard error.
func (*AudioRouter) Publish ¶
func (r *AudioRouter) Publish(frame Frame)
Publish enqueues a frame for fan-out. Non-blocking by contract: drops the frame silently if the router input buffer is full or the router has been closed. The non-blocking behaviour is load-bearing — see the observer-model doc in types.go. Do not change this to block.
Safe to call after Close: the closed flag is checked under a read lock that excludes the actual channel close, so we never reach a send to a closed channel and never panic.
func (*AudioRouter) RegisterDrainHandler ¶
func (r *AudioRouter) RegisterDrainHandler(h DrainHandler)
RegisterDrainHandler adds a callback that the turn loop can wait on between turns via WaitOutputDrained. Used by subscribers that physically play audio (LocalSink) so the next user turn doesn't start while the previous assistant audio is still audibly playing.
Handlers run only when an external controller calls WaitOutputDrained. They never run on the dispatch goroutine and never affect Publish — the data path stays observer-style.
Multiple handlers are supported (one router can drive several local playback paths in theory). They run concurrently; WaitOutputDrained returns when all have completed or the deadline elapses.
func (*AudioRouter) Subscribe ¶
func (r *AudioRouter) Subscribe(id string, bufSize int) <-chan Frame
Subscribe registers a new consumer. Returns a receive-only channel. bufSize controls how many frames may queue before drop-on-overflow.
A duplicate id is treated as an overwrite: the previously-registered channel is closed (so any goroutine reading it observes EOF, same as Unsubscribe) and replaced with the new one. A duplicate is logged because it usually indicates a bug at the call site.
Subscribe-after-Close returns an already-closed channel rather than registering a new consumer that would never be serviced (the dispatch goroutine has exited, so any frames published after Close are dropped — see Publish). The caller's range loop on the returned channel terminates immediately, which is what they would observe if they had subscribed before Close and the router had since shut down.
func (*AudioRouter) Unsubscribe ¶
func (r *AudioRouter) Unsubscribe(id string)
Unsubscribe removes a consumer and closes its channel. Subsequent Publish calls will not deliver to this consumer.
func (*AudioRouter) WaitOutputDrained ¶
func (r *AudioRouter) WaitOutputDrained(ctx context.Context)
WaitOutputDrained blocks until every registered DrainHandler has returned, or until ctx is canceled. Each handler is invoked with the same ctx; they're expected to apply their own timeout if needed.
Safe to call when no handlers are registered (returns immediately). Safe to call after Close — handlers are still tracked.
type Direction ¶
type Direction string
Direction identifies whether audio came from the user (input) or the agent (output).
type DrainHandler ¶
DrainHandler is an optional callback a subscriber can register with AudioRouter.RegisterDrainHandler to participate in the inter-turn drain barrier. The handler should block until its internal queues are empty or the deadline implied by ctx elapses, then return.
type Frame ¶
Frame is the canonical audio chunk format used inside the router. Samples are mono s16le at the canonical rate (configured per-run).
type LocalSink ¶
type LocalSink struct {
// contains filtered or unexported fields
}
LocalSink plays Frame stereo (user L / agent R) on the host audio device via oto/v3.
LocalSink is an OBSERVER, not part of the data path. It subscribes to audio coming off the AudioRouter; Push is fire-and-forget and drops on full buffer. That drop semantics is load-bearing — see the observer-model doc in types.go. The sink cannot be allowed to push back on its producer because the same audio is fanned out to the duplex provider's session, and provider VAD timing must not be warped by who happens to be listening locally (often nobody, in parallel CI).
On any oto failure (no audio device, headless Linux without ALSA/Pulse) the sink falls back to no-op — the run continues, only local playback is missing. SSE / TUI monitoring surfaces remain operational regardless.
Lifecycle: NewLocalSink either opens an oto context and creates a Player fed by an internal io.Reader, or returns a noop sink. Push enqueues mono frames per direction; the io.Reader interleaves them into stereo s16le on demand from oto's audio thread. Close releases the player.
func NewLocalSink ¶
func NewLocalSink(cfg LocalSinkConfig) (*LocalSink, error)
NewLocalSink constructs a sink. On failure to open the audio device, returns a no-op sink and a nil error — local playback is best-effort.
When cfg.Headless is true, oto is bypassed entirely and a ticker goroutine drains the streamReader at audio rate. RMS publishing and capture still work; nothing actually reaches host speakers. Useful for CI / bash invocations where there's no audio device.
func (*LocalSink) Push ¶
Push enqueues a frame for playback. Non-blocking by contract; drops on full buffer. The drop semantics is load-bearing — this sink is an observer on a fan-out bus that also feeds the duplex provider's session, so blocking here would warp the provider's VAD timing via the broadcast. See the observer-model doc in types.go before considering any change to this. In no-op mode Push is an immediate return.
func (*LocalSink) WaitDrained ¶
WaitDrained blocks until both per-direction audio queues are empty, or until the deadline elapses, then waits a small grace period for the host audio device's own output buffer to flush.
LocalSink remains a passive observer — this method exists so a controller (the duplex executor's turn loop) can opt-in to "wait for local playback to finish before starting the next turn." The pipeline's data path doesn't change; nobody pushes back on the producer.
No-op when the sink is in noop mode (no audio device).
type LocalSinkConfig ¶
type LocalSinkConfig struct {
// Rate is the canonical sample rate for the stereo output stream.
// If 0, defaults to Rate24k.
Rate int
// RMSPublisher, if non-nil, is called from the audio thread at ~30 Hz
// with the RMS levels of the samples just sent to the audio device. This
// is what drives the level meter — values reflect what is *playing*,
// not what is queued, so the meter stays in sync with the user's ears.
RMSPublisher func(RMSFrame)
// CapturePath, when non-empty, names a file the sink writes every byte
// oto (or the headless pull goroutine) pulls into. Format: interleaved
// s16le stereo at Rate. Useful for offline analysis of what the audio
// device saw without involving the host audio stack.
CapturePath string
// Headless skips opening an oto context and instead drives Read from a
// software ticker at audio rate. Useful for CI / bash invocation and
// tests where there's no audio device — RMS publishing and capture
// still work, but nothing actually plays through speakers.
Headless bool
// contains filtered or unexported fields
}
LocalSinkConfig configures the host-audio playback sink.
type Monitor ¶
type Monitor struct {
// contains filtered or unexported fields
}
Monitor owns the process-wide host-audio playback path and routes one run's audio to it at a time. It exists because oto/v3's audio context is a process-wide singleton — concurrent runs cannot each open their own LocalSink. The Monitor keeps a single sink alive, holds a registry of per-run AudioRouters, and forwards frames from the *active* run to the sink. Callers switch between runs with SetActiveRun.
RMS frames originate from the sink (driven by oto's pull cadence so the meter follows playback timing) and are fanned out to subscribers via SubscribeRMS — independent of which run is active.
Lifecycle: NewMonitor opens the sink eagerly (or returns a no-sink monitor if oto fails). Close stops forwarding and releases the sink. AttachRouter / DetachRouter are paired with each run's lifetime; the engine doesn't have to know about the Monitor at all — the CLI wires it via the engine's AudioMonitorHook.
func NewMonitor ¶
func NewMonitor(cfg MonitorConfig) (*Monitor, error)
NewMonitor constructs a Monitor. When EnableLocalSink is true and the host audio device is available, host playback is enabled; otherwise the monitor runs in metering-only mode and never produces sound.
func (*Monitor) ActiveRunID ¶
ActiveRunID returns the run currently routed to the sink, or "" if nothing is active.
func (*Monitor) AttachRouter ¶
func (m *Monitor) AttachRouter(runID string, router *AudioRouter)
AttachRouter registers a per-run AudioRouter with the monitor. If no run is currently active for audio, this run becomes active immediately (so a single-run TTY session "just works"). For concurrent runs only the first auto-activates; subsequent runs sit in the registry until the user picks them via SetActiveRun.
The router self-removes from the registry when it closes (the engine closes per-run routers at run end), so callers don't need to pair this with DetachRouter unless they want to detach early.
func (*Monitor) Close ¶
func (m *Monitor) Close()
Close stops forwarding and releases the sink. Safe to call multiple times. After Close, AttachRouter / SetActiveRun become no-ops.
func (*Monitor) DetachRouter ¶
DetachRouter removes a run's AudioRouter from the monitor. If that run was the active audio source, the monitor falls back to any other registered router (so listening "moves on" automatically when the run you were listening to ends and another is still going).
func (*Monitor) SetActiveRun ¶
SetActiveRun switches host playback (and the meter) to the named run. Returns false when the run isn't registered with the monitor — callers can use this to know whether the switch took effect.
func (*Monitor) SubscribeRMS ¶
SubscribeRMS returns a channel of RMS frames sourced from the sink's playback timing (so the meter is in sync with the user's ears, regardless of which run is active). The buffer is per-subscriber; frames overflow drop silently for that subscriber only.
type MonitorConfig ¶
type MonitorConfig struct {
// Rate is the canonical sample rate for playback. Must match what
// per-run AudioRouters publish at; both sides resample to this rate.
Rate int
// EnableLocalSink toggles host-audio playback. When false the monitor
// still distributes RMS frames (useful for tests / headless runs that
// want the meter without sound).
EnableLocalSink bool
// Headless skips opening an audio device but still drives the sink at
// audio rate via a software ticker. Combined with CapturePath it lets
// CI / bash invocations produce sample-accurate capture files without
// a host audio interface.
Headless bool
// CapturePath, when non-empty, mirrors every byte the sink pulled to
// a file at this path. Format: interleaved s16le stereo at Rate.
// Useful for debugging "is my audio data choppy or is the audio
// device just misbehaving?" — play the file with ffplay/sox to hear
// exactly what the audio thread saw.
CapturePath string
}
MonitorConfig configures a Monitor.
type MonitorMode ¶
type MonitorMode string
MonitorMode controls when audio monitoring is active.
const ( // ModeAuto enables monitoring when a TTY is attached and a duplex // scenario is running, unless --ci is set. ModeAuto MonitorMode = "auto" // ModeOn forces monitoring on regardless of TTY/CI state. ModeOn MonitorMode = "on" // ModeOff disables monitoring; MonitorTap is not added to the pipeline. ModeOff MonitorMode = "off" )
type MonitorTap ¶
MonitorTap is a pipeline stage that observes audio elements and forwards them to the AudioRouter. Pure passthrough — elements flow to downstream stages unmodified. Router publishing is non-blocking by contract, so a slow router or full consumer buffer never affects pipeline throughput.
This is the *boundary* between the data path (Stage chain) and the observer side (router + LocalSink + RMS subscribers + ...). Audio flowing through here goes two places at once: forward to the next pipeline stage (data path), and sideways onto the router (observers). The observer side must not influence the data path — see the observer-model doc in types.go. If a sink can't keep up, frames are dropped at the router or sink, never propagated back as backpressure.
If observer-side smooth playback matters (someone is listening live to a bursty mock provider), the answer is to add another AudioPacingStage on the data path upstream of this tap, not to make the observer side blocking.
func NewMonitorTap ¶
func NewMonitorTap(router *AudioRouter, config MonitorTapConfig) *MonitorTap
NewMonitorTap constructs a tap that publishes audio elements to the given router.
func (*MonitorTap) Process ¶
func (t *MonitorTap) Process( ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement, ) error
Process implements stage.Stage. It taps audio elements to the router and passes every element through to the output channel unchanged.
type MonitorTapConfig ¶
type MonitorTapConfig struct {
// Position determines whether tapped audio is treated as input
// (user/selfplay) or output (agent).
Position stage.RecordingPosition
}
MonitorTapConfig configures the MonitorTap stage.
type Options ¶
type Options struct {
// Mode controls when the MonitorTap is added to the pipeline.
Mode MonitorMode
// Rate is the canonical sample rate inside the router. Must be one
// of Rate16k, Rate24k, or Rate48k.
Rate int
// LocalSink enables oto-backed local playback when monitoring is on.
LocalSink bool
// SSEPlayback enables the web SSE relay subscription.
SSEPlayback bool
// LevelMeter enables the RMSFrame channel for TUI consumption.
LevelMeter bool
}
Options configures the audio monitoring stack at engine startup.
func DefaultOptions ¶
func DefaultOptions() Options
DefaultOptions returns sensible defaults: auto mode, 24 kHz, all surfaces enabled.
Source Files