feat(agent): stream package operation output over per-operation event channel

[Vladyslav Nikonov (vnikonov-devolutions)]

Clients of the Devolutions Agent package broker now receive live stdout/stderr output and instant status-change notifications while package operations run.

For each executed operation, the broker opens a dedicated per-operation event channel (a local named pipe advertised in the operation submission) implementing the event-channel protocol v1.0. Connected clients get a HELLO frame, real-time STDOUT/STDERR data frames (when output capture is requested), a STATUS_UPDATED notification on every status transition, and a FINISH frame when the operation completes.

This restores the output visibility that was removed by the API v0.3 upgrade and improves on it: instead of post-hoc capture returned after the operation ends, output now streams in real time. Slow or absent clients never stall or fail the operation.

Stacked on #1913 (operation cancellation); base branch is vnikonov-devolutions-dgw-agent-broker-cancellation.

Issue: DGW-438

[Vladyslav Nikonov (vnikonov-devolutions)]

Implementation notes

Architecture (devolutions-agent/src/broker/event_channel.rs)

• Utf8StreamChunker: splits raw output bytes into frames of at most 64 KiB (MAX_EVENT_FRAME_BODY_BYTES) without ever splitting a multi-byte UTF-8 character across frames (protocol guarantee). Incomplete trailing sequences are buffered; invalid bytes become U+FFFD.
• OperationEventSink: cloneable, non-blocking handle used by producer threads (executor reader threads, tracker). Bounded in-memory queue with a 256 KiB per-stream budget; when a slow/absent client lets the queue fill, chunks are dropped whole and accounted, then reported via StdoutOverflow/StderrOverflow{bytes_skipped} frames. Producers never block, so the operation is never stalled by the channel.
• Windows pipe server task: created before execute() returns, so the descriptor in the OperationSubmission is always immediately connectable.

Pipe security

• DACL: SYSTEM and Administrators get full control; the requesting client's SID gets FILE_GENERIC_READ only. Not world-accessible.
• first_pipe_instance + max_instances(1), outbound-only (server→client), single client, one-shot (no reconnects, per protocol).

Interop

• The EventChannel.path descriptor carries the bare pipe name Devolutions.Now.PackageBroker.Operation.<operation-id> (no \\.\pipe\ prefix), matching the .NET client and upstream samples.

Frame lifecycle

• HELLO (v1.0) first on connect → STDOUT/STDERR data frames (only when CaptureOutput=true) and STATUS_UPDATED on every transition (Running, Canceling, terminal — always sent regardless of CaptureOutput) → FINISH last, then drain and close.

Executor changes

• The single combined output pipe is split into separate stdout and stderr anonymous pipes, each with a dedicated reader thread forwarding raw chunks to the sink. Pre-operation command output is streamed too (honors CaptureOutput); kill-before/post commands are not captured. The dry-run executor emits its simulated line through the sink.

Degradation

• Channel creation is best-effort: on failure the broker logs a warning and returns event_channel: None — the operation itself proceeds normally. Idempotent resubmission returns the stored descriptor.

Tunables

• 60 s linger for never-connected clients after the operation finishes; 30 s client-drain timeout after FINISH is written (lets the client consume buffered frames before the handle closes).

Tests

• 9 unit tests: chunker UTF-8 boundary handling, overflow accounting, byte-conservation invariant (received + skipped == pushed).
• 4 pipe-level tests with a real named-pipe client.
• 3 server end-to-end tests: HELLO first + streamed stdout + STATUS_UPDATED + FINISH; CaptureOutput=false yields status frames but no data frames; operation completes with no client connected.

Validation

• cargo +nightly fmt --all (clean), cargo clippy --workspace --tests -- -D warnings (clean), cargo test --workspace (green; one unrelated flaky testsuite integration test passed on rerun).

[Copilot]

Pull request overview

Note

Copilot was unable to run its full agentic suite in this review.

Adds a per-operation event channel that clients can connect to (Windows named pipe) to receive status transitions and optionally streamed stdout/stderr.

Changes:

• Open and advertise an event channel on operation submission; reuse the same descriptor for idempotent resubmissions.
• Track and emit event frames on status transitions and operation completion.
• Stream captured stdout/stderr chunks to the event channel while still retaining a bounded diagnostic tail in ExecutionOutput.

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
devolutions-agent/src/broker/server/mod.rs	Opens per-operation event channels on Windows and adds event-channel-focused tests.
devolutions-agent/src/broker/operation_tracker.rs	Stores event sink/descriptor per operation and emits frames on status changes and terminal completion.
devolutions-agent/src/broker/mod.rs	Exposes the new event_channel module.
devolutions-agent/src/broker/executor/windows/process.rs	Splits stdout/stderr capture and forwards raw chunks to the event sink while keeping a bounded tail.
devolutions-agent/src/broker/executor/windows/mod.rs	Updates create_process calls to use OutputCapture and pass the event sink.
devolutions-agent/src/broker/executor/mod.rs	Adds event_sink to ExecutionContext and emits dry-run output via the sink when enabled.
devolutions-agent/src/broker/event_channel.rs	Implements the event channel queue/sink and Windows named-pipe writer plus unit/integration tests.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 7 out of 7 changed files in this pull request and generated 1 comment.

Suppressed comments (2)

devolutions-agent/src/broker/server/mod.rs:272

• register publishes the operation with event_channel: None before this lookup and the later pipe creation/update. Because the broker serves connections concurrently (broker/pipe.rs:75-101), a simultaneous idempotent retry can observe the indexed operation in this initialization window and return event_channel: None, even though the first request installs the channel moments later. Make channel initialization part of the atomic registration path, or represent initialization explicitly and wait for it before returning a retry response.

            let mut event_channel = self
                .tracker
                .get(&operation_id)
                .and_then(|operation| operation.event_channel);

devolutions-agent/src/broker/operation_tracker.rs:235

• Repeated cancellation requests while the operation is already Canceling enqueue another StatusUpdated even though no status transition occurred. Besides producing spurious protocol notifications, these control frames are not covered by the output byte budget, so idempotent cancel traffic can grow the supposedly bounded queue. Notify only when the status actually changes to Canceling.

            if let Some(sink) = &op.event_sink {
                sink.status_updated();
            }

[Copilot]

Pull request overview

Copilot reviewed 7 out of 7 changed files in this pull request and generated 1 comment.

Suppressed comments (3)

devolutions-agent/src/broker/event_channel.rs:500

• A connected client that stops reading can leave write_all pending indefinitely once the pipe buffer fills. The no-client and post-finish paths are timed out, but this path is not, so each such connection can retain its writer task, pipe handle, and buffered queue forever; repeated operations can exhaust service resources. Apply a bounded write/session timeout and abandon the sink when it expires.

        while let Some(frame) = sink.next_frame().await {
            if let Err(error) = write_frame(&mut server, &frame).await {

devolutions-agent/src/broker/operation_tracker.rs:161

• A cancellation can race between run_plan's initial token check and the process-started callback. If the tracker is already Canceling, this method overwrites it with Running and emits a misleading status notification before the terminal update. Record started_at, but only transition and notify when the current status is Starting.

            if let Some(sink) = &op.event_sink {
                sink.status_updated();

devolutions-agent/src/broker/operation_tracker.rs:235

• Canceling is non-terminal, so every idempotent cancel retry enters this block and emits another StatusUpdated even though no transition occurred. Besides violating the channel's transition semantics, these unbudgeted control frames can grow the queue under repeated retries. Only perform the transition and notification when the operation is not already Canceling.

            if let Some(sink) = &op.event_sink {
                sink.status_updated();
            }