ChannelProgressReporter
NexusLabs.Foundry.MicrosoftAgentFramework¶
NexusLabs.Foundry.MicrosoftAgentFramework.Progress¶
ChannelProgressReporter Class¶
Bounded, backpressured IProgressReporter that pushes events to a System.Threading.Channels.Channel<> and drains them to sinks on a background task. Use this when sinks do I/O (database, network) and you want production decoupled from consumption while capacity is available, without letting an unbounded backlog of undelivered events accumulate in memory.
public sealed class ChannelProgressReporter : NexusLabs.Foundry.MicrosoftAgentFramework.Progress.IProgressReporter, System.IAsyncDisposable
Inheritance System.Object 🡒 ChannelProgressReporter
Implements IProgressReporter, System.IAsyncDisposable
Remarks¶
Report(IProgressEvent) enqueues to the channel via System.Threading.Channels.ChannelWriter<>.WriteAsync(<>.Threading.CancellationToken)
(Wait mode — never TryWrite, so a full channel never silently drops an event).
This is not fully non-blocking: whenever capacity is available — the
common case — the enqueue completes synchronously and Report(IProgressEvent) returns
immediately. Only when the channel is momentarily saturated (the background consumer is
still busy with a slow sink and the buffer is full) does Report(IProgressEvent) itself
synchronously wait for System.Threading.Channels.ChannelWriter<>.WriteAsync(<>.Threading.CancellationToken) to complete, applying
backpressure directly to the caller instead of accumulating an unbounded set of
fire-and-forget pending-write tasks. A background consumer drains events to all sinks in
channel order. Concurrent writers may interleave; each event's
SequenceNumber remains the authoritative global ordering key.
A System.Threading.Channels.ChannelClosedException from enqueueing after the channel has already been completed (e.g. after DisposeAsync()) — the one specific, expected failure mode for a saturated System.Threading.Channels.ChannelWriter<>.WriteAsync(<>.Threading.CancellationToken) wait — is surfaced to the configured IProgressReporterErrorHandler (once per registered sink) rather than thrown out of Report(IProgressEvent), mirroring how a sink exception during consumption is reported. Any other, unexpected exception from that wait propagates directly out of Report(IProgressEvent) instead of being reshaped into a handled error.
CreateChild(string) returns a lightweight wrapper that shares the parent's channel — no additional background tasks are created.
Call DisposeAsync() to complete the writer and drain the background consumer. Because Report(IProgressEvent) never leaves an enqueue running in the background unobserved — it either completes synchronously or the caller's own call is the one waiting on it — there is no separate pending-write set for disposal to wait on: completing the writer and awaiting the consumer task is sufficient.
Constructors¶
ChannelProgressReporter(string, IReadOnlyList<IProgressSink>, IProgressSequence, IProgressReporterErrorHandler, string, string, int, int) Constructor¶
Creates a channel-based reporter with the given sinks. Starts a background consumer immediately.
public ChannelProgressReporter(string workflowId, System.Collections.Generic.IReadOnlyList<NexusLabs.Foundry.MicrosoftAgentFramework.Progress.IProgressSink> sinks, NexusLabs.Foundry.MicrosoftAgentFramework.Progress.IProgressSequence sequence, NexusLabs.Foundry.MicrosoftAgentFramework.Progress.IProgressReporterErrorHandler? errorHandler=null, string? agentId=null, string? parentAgentId=null, int depth=0, int capacity=1000);
Parameters¶
workflowId System.String
sinks System.Collections.Generic.IReadOnlyList<IProgressSink>
sequence IProgressSequence
errorHandler IProgressReporterErrorHandler
agentId System.String
parentAgentId System.String
depth System.Int32
capacity System.Int32
Properties¶
ChannelProgressReporter.AgentId Property¶
Gets the current agent ID, or null for workflow-level scope.
Implements AgentId
Property Value¶
ChannelProgressReporter.Depth Property¶
Gets the nesting depth.
Implements Depth
Property Value¶
ChannelProgressReporter.WorkflowId Property¶
Gets the workflow ID for this reporter's scope.
Implements WorkflowId
Property Value¶
Methods¶
ChannelProgressReporter.CreateChild(string) Method¶
Creates a child reporter scoped to a specific agent. Events emitted by the child
carry the parent's agent ID as ParentAgentId and an incremented Depth.
public NexusLabs.Foundry.MicrosoftAgentFramework.Progress.IProgressReporter CreateChild(string agentId);
Parameters¶
agentId System.String
The agent ID for the child scope.
Implements CreateChild(string)
Returns¶
ChannelProgressReporter.DisposeAsync() Method¶
Completes the channel and waits for the background consumer to drain all remaining events to sinks.
Implements DisposeAsync()
Returns¶
System.Threading.Tasks.ValueTask
Remarks¶
No in-flight enqueue can ever be orphaned by completing the channel here: every Report(IProgressEvent) call that observed a saturated channel is, by construction, still synchronously blocked inside that same call waiting on its own System.Threading.Channels.ChannelWriter<>.WriteAsync(<>.Threading.CancellationToken) — there is no separate background task whose completion this method would otherwise need to await first.
ChannelProgressReporter.NextSequence() Method¶
Allocates the next globally-ordered sequence number.
Implements NextSequence()
Returns¶
ChannelProgressReporter.Report(IProgressEvent) Method¶
Emits a progress event to all registered sinks.
public void Report(NexusLabs.Foundry.MicrosoftAgentFramework.Progress.IProgressEvent progressEvent);
Parameters¶
progressEvent IProgressEvent
Implements Report(IProgressEvent)