diff --git a/Cargo.lock b/Cargo.lock
index b8d136d..0c74f63 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -109,12 +109,6 @@ dependencies = [
"tower-service",
]
-[[package]]
-name = "base64"
-version = "0.22.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
-
[[package]]
name = "bitflags"
version = "2.13.1"
@@ -634,7 +628,6 @@ dependencies = [
name = "now-policy-api"
version = "0.2.0"
dependencies = [
- "base64",
"chrono",
"derive_more",
"now-policy",
diff --git a/policies/docs/event-channel-protocol.md b/policies/docs/event-channel-protocol.md
new file mode 100644
index 0000000..ce8d17c
--- /dev/null
+++ b/policies/docs/event-channel-protocol.md
@@ -0,0 +1,171 @@
+# Package broker event channel protocol
+
+Version: **1.0**
+
+This document specifies the `NOW_BROKER` frame protocol carried over a
+per-operation event channel between a Devolutions NOW package broker (server)
+and a broker client.
+
+Reference implementations:
+
+- Rust: `policies/rust/now-policy-api/src/event_channel.rs`
+- .NET: `policies/dotnet/Devolutions.Now.Policy.Api/EventChannel.cs`
+- Shared test fixture: `policies/rust/now-policy-server-template/assets/samples/frames/event-channel.frames.bin`
+
+## Purpose
+
+For each executed operation the broker (when it supports event channels) opens
+a dedicated event channel and returns its descriptor in the
+`ExecutionResponse` (`Operation.EventChannel`):
+
+```json
+"EventChannel": {
+ "Kind": "LocalPipe",
+ "Path": "Devolutions.Now.PackageBroker.Operation.op-000001"
+}
+```
+
+`Kind` is extendable; `LocalPipe` (a local named pipe) is the only transport
+currently defined. `Path` is the transport-specific address the client
+connects to.
+
+The channel is used to:
+
+- push both stdout and stderr data over a single channel, greatly simplifying
+ client code and preserving the correct sequential order of interleaved
+ stdout/stderr output — output data frames are only sent when the execute
+ request opted in via `CaptureOutput`;
+- eliminate periodic status polling: the client sends `StatusRequest` HTTP
+ queries only when notified that something actually changed;
+- enforce correct UTF-8 character boundaries mid-transfer.
+
+## Transport properties
+
+- **One-way**: read-only from the client side. The client never writes.
+- **Minimal overhead**: fixed 6-byte header, no encoding of payload bytes.
+- **Extendable**: new frame kinds may be added down the line; decoders MUST
+ ignore frames with unknown kinds.
+
+## Frame layout
+
+All integers are **little-endian**.
+
+```text
+NOW_BROKER_FRAME
+| u32 frame_size | u16 frame_kind | [frame_size; u8] frame_body |
+```
+
+- `frame_size` is the body length in bytes; it excludes the 6-byte header.
+- `frame_size` MUST NOT exceed **65536** (64 KiB). Producers MUST split larger
+ output into multiple frames before encoding (the reference `Encode`
+ implementations reject oversized bodies rather than splitting them);
+ decoders MUST treat a larger value as a fatal
+ protocol error and close the channel.
+- A decode error (oversized frame, malformed body, invalid UTF-8) is not
+ recoverable: the client SHOULD close the channel and fall back to
+ HTTP status queries.
+- End-of-stream in the middle of a frame means the stream was truncated
+ (e.g. the broker crashed or the pipe broke); clients SHOULD treat this as
+ an error rather than a graceful close.
+
+## Frame kinds
+
+| Kind | Name | Body size | Direction |
+| -------- | --------------------------- | --------- | --------------- |
+| `0x0000` | `NOW_BROKER_HELLO` | 4 | server → client |
+| `0x0001` | `NOW_BROKER_STATUS_UPDATED` | 0 | server → client |
+| `0x0002` | `NOW_BROKER_FINISH` | 0 | server → client |
+| `0x0003` | `NOW_BROKER_STDOUT` | variable | server → client |
+| `0x0004` | `NOW_BROKER_STDERR` | variable | server → client |
+| `0x0005` | `NOW_BROKER_STDOUT_OVERFLOW`| 4 | server → client |
+| `0x0006` | `NOW_BROKER_STDERR_OVERFLOW`| 4 | server → client |
+
+### NOW_BROKER_HELLO (0x0000)
+
+Sent as the first frame on the channel; acknowledges to the client that the
+transport is ready and advertises the protocol version.
+
+```text
+| frame_size = 4 | frame_kind = 0x0000 | u16 version_major = 1 | u16 version_minor = 0 |
+```
+
+Clients MUST reject channels whose `version_major` they do not support.
+`version_minor` increments are backward compatible (new frame kinds only).
+
+### NOW_BROKER_STATUS_UPDATED (0x0001)
+
+Sent when the operation status has changed and awaits being queried via a
+`StatusRequest` HTTP query. This frame deliberately omits any status
+information: complex data is queried over the main HTTP (pipe) API.
+
+```text
+| frame_size = 0 | frame_kind = 0x0001 |
+```
+
+### NOW_BROKER_FINISH (0x0002)
+
+Sent when the operation is finished; the client should call `StatusRequest` to
+query more info. The channel can be gracefully closed by the client after this
+frame. No further frames follow.
+
+```text
+| frame_size = 0 | frame_kind = 0x0002 |
+```
+
+### NOW_BROKER_STDOUT (0x0003)
+
+Sent when new stdout data is available. The body is UTF-8 encoded data;
+character boundaries are guaranteed by the broker side — a multi-byte UTF-8
+character is never split across frames.
+
+```text
+| frame_size = variable | frame_kind = 0x0003 | [frame_size; u8] data |
+```
+
+### NOW_BROKER_STDERR (0x0004)
+
+Same as `NOW_BROKER_STDOUT` but for stderr.
+
+```text
+| frame_size = variable | frame_kind = 0x0004 | [frame_size; u8] data |
+```
+
+### NOW_BROKER_STDOUT_OVERFLOW (0x0005)
+
+Sent when the client was too slow to read stdout and some data was truncated.
+
+```text
+| frame_size = 4 | frame_kind = 0x0005 | u32 bytes_skipped |
+```
+
+### NOW_BROKER_STDERR_OVERFLOW (0x0006)
+
+Same as `NOW_BROKER_STDOUT_OVERFLOW` but for stderr.
+
+```text
+| frame_size = 4 | frame_kind = 0x0006 | u32 bytes_skipped |
+```
+
+## Typical session
+
+```text
+server → HELLO (1.0)
+server → STDOUT "Resolving package…\n"
+server → STDOUT "Downloading…\n"
+server → STATUS_UPDATED (client issues StatusRequest over HTTP)
+server → STDERR "warning: …\n"
+server → STDOUT_OVERFLOW 4096 (client was too slow; 4096 bytes lost)
+server → STDOUT "Installed.\n"
+server → STATUS_UPDATED
+server → FINISH (client issues final StatusRequest, closes pipe)
+```
+
+## Versioning and extension rules
+
+- New frame kinds are added with new `frame_kind` values and a
+ `version_minor` bump; decoders MUST skip unknown kinds (the header is
+ sufficient to do so).
+- Changing the layout of an existing frame kind requires a `version_major`
+ bump.
+- The 64 KiB body limit is part of the protocol contract and does not change
+ within major version 1.
diff --git a/policies/dotnet/Devolutions.Now.Policy.Api/Enums.cs b/policies/dotnet/Devolutions.Now.Policy.Api/Enums.cs
index 524fee4..4959a6c 100644
--- a/policies/dotnet/Devolutions.Now.Policy.Api/Enums.cs
+++ b/policies/dotnet/Devolutions.Now.Policy.Api/Enums.cs
@@ -114,4 +114,12 @@ public enum ErrorCode
BrokerPaused,
InternalError,
Timeout,
+}
+
+/// Transport kind of a per-operation event channel.
+[JsonConverter(typeof(JsonStringEnumConverter))]
+public enum EventChannelKind
+{
+ /// Local named pipe carrying NOW_BROKER event frames.
+ LocalPipe,
}
\ No newline at end of file
diff --git a/policies/dotnet/Devolutions.Now.Policy.Api/EventChannel.cs b/policies/dotnet/Devolutions.Now.Policy.Api/EventChannel.cs
new file mode 100644
index 0000000..c9d2b3f
--- /dev/null
+++ b/policies/dotnet/Devolutions.Now.Policy.Api/EventChannel.cs
@@ -0,0 +1,335 @@
+using System.Buffers.Binary;
+using System.Text;
+using System.Text.Json.Serialization;
+
+namespace Devolutions.Now.Policy.Api;
+
+/// Descriptor of a per-operation event channel returned in the execution response.
+///
+/// The channel is one-way (read-only from the client side) and carries the
+/// NOW_BROKER event frame protocol: stdout/stderr data and status change
+/// notifications. See policies/docs/event-channel-protocol.md.
+///
+public sealed class EventChannel
+{
+ /// Transport kind of the channel.
+ [JsonPropertyName("Kind")]
+ public EventChannelKind Kind { get; set; }
+
+ ///
+ /// Transport-specific path. For this is the
+ /// pipe name/path the client should connect to.
+ ///
+ [JsonPropertyName("Path")]
+ public string Path { get; set; } = "";
+}
+
+/// Constants of the NOW_BROKER event channel frame protocol.
+public static class EventChannelProtocol
+{
+ /// Protocol major version advertised in the Hello frame.
+ public const ushort VersionMajor = 1;
+
+ /// Protocol minor version advertised in the Hello frame.
+ public const ushort VersionMinor = 0;
+
+ /// Size in bytes of the fixed frame header (u32 frame_size + u16 frame_kind).
+ public const int FrameHeaderSize = 6;
+
+ ///
+ /// Maximum allowed frame body size in bytes. Protects decoders from unbounded
+ /// buffering; encoders must split larger output into multiple frames.
+ ///
+ public const int MaxFrameBodyBytes = 65536;
+}
+
+/// NOW_BROKER event frame kind discriminators.
+public enum EventFrameKind : ushort
+{
+ /// NOW_BROKER_HELLO
+ Hello = 0x0000,
+
+ /// NOW_BROKER_STATUS_UPDATED
+ StatusUpdated = 0x0001,
+
+ /// NOW_BROKER_FINISH
+ Finish = 0x0002,
+
+ /// NOW_BROKER_STDOUT
+ Stdout = 0x0003,
+
+ /// NOW_BROKER_STDERR
+ Stderr = 0x0004,
+
+ /// NOW_BROKER_STDOUT_OVERFLOW
+ StdoutOverflow = 0x0005,
+
+ /// NOW_BROKER_STDERR_OVERFLOW
+ StderrOverflow = 0x0006,
+}
+
+/// A decoded NOW_BROKER event frame.
+public abstract class EventFrame
+{
+ private protected EventFrame()
+ {
+ }
+
+ /// Frame kind discriminator for this frame.
+ public abstract ushort Kind { get; }
+
+ ///
+ /// First frame on the channel; acknowledges that the transport is ready and
+ /// advertises the protocol version.
+ ///
+ public sealed class Hello(ushort versionMajor, ushort versionMinor) : EventFrame
+ {
+ public ushort VersionMajor { get; } = versionMajor;
+
+ public ushort VersionMinor { get; } = versionMinor;
+
+ public override ushort Kind => (ushort)EventFrameKind.Hello;
+ }
+
+ ///
+ /// The operation status changed and awaits a StatusRequest HTTP query.
+ /// Deliberately carries no payload: complex data is queried over HTTP.
+ ///
+ public sealed class StatusUpdated : EventFrame
+ {
+ public override ushort Kind => (ushort)EventFrameKind.StatusUpdated;
+ }
+
+ /// New stdout data. Strictly UTF-8; character boundaries are guaranteed by the broker.
+ public sealed class Stdout(string data) : EventFrame
+ {
+ public string Data { get; } = data;
+
+ public override ushort Kind => (ushort)EventFrameKind.Stdout;
+ }
+
+ /// New stderr data. Strictly UTF-8; character boundaries are guaranteed by the broker.
+ public sealed class Stderr(string data) : EventFrame
+ {
+ public string Data { get; } = data;
+
+ public override ushort Kind => (ushort)EventFrameKind.Stderr;
+ }
+
+ ///
+ /// The operation finished; query StatusRequest for details. The channel can be
+ /// gracefully closed by the client after this frame.
+ ///
+ public sealed class Finish : EventFrame
+ {
+ public override ushort Kind => (ushort)EventFrameKind.Finish;
+ }
+
+ /// The client was too slow to read stdout and some data was truncated.
+ public sealed class StdoutOverflow(uint bytesSkipped) : EventFrame
+ {
+ public uint BytesSkipped { get; } = bytesSkipped;
+
+ public override ushort Kind => (ushort)EventFrameKind.StdoutOverflow;
+ }
+
+ /// The client was too slow to read stderr and some data was truncated.
+ public sealed class StderrOverflow(uint bytesSkipped) : EventFrame
+ {
+ public uint BytesSkipped { get; } = bytesSkipped;
+
+ public override ushort Kind => (ushort)EventFrameKind.StderrOverflow;
+ }
+
+ ///
+ /// A frame with an unknown kind. Must be ignored by consumers to allow
+ /// forward-compatible protocol extension.
+ ///
+ public sealed class Unknown(ushort kind, byte[] body) : EventFrame
+ {
+ public byte[] Body { get; } = body;
+
+ public override ushort Kind { get; } = kind;
+ }
+
+ /// Encode the frame (header + body) into a byte array.
+ /// The frame body exceeds .
+ public byte[] Encode()
+ {
+ var body = this switch
+ {
+ Hello hello => EncodeHelloBody(hello),
+ StatusUpdated or Finish => [],
+ Stdout stdout => Encoding.UTF8.GetBytes(stdout.Data),
+ Stderr stderr => Encoding.UTF8.GetBytes(stderr.Data),
+ StdoutOverflow overflow => EncodeOverflowBody(overflow.BytesSkipped),
+ StderrOverflow overflow => EncodeOverflowBody(overflow.BytesSkipped),
+ Unknown unknown => unknown.Body,
+ _ => throw new EventFrameException($"Unsupported frame type {GetType().Name}."),
+ };
+
+ if (body.Length > EventChannelProtocol.MaxFrameBodyBytes)
+ {
+ throw new EventFrameException(
+ $"Frame body size {body.Length} exceeds the maximum of {EventChannelProtocol.MaxFrameBodyBytes} bytes.");
+ }
+
+ var frame = new byte[EventChannelProtocol.FrameHeaderSize + body.Length];
+ BinaryPrimitives.WriteUInt32LittleEndian(frame, (uint)body.Length);
+ BinaryPrimitives.WriteUInt16LittleEndian(frame.AsSpan(4), Kind);
+ body.CopyTo(frame.AsSpan(EventChannelProtocol.FrameHeaderSize));
+ return frame;
+ }
+
+ /// Decode a frame from a kind discriminator and body bytes.
+ /// The body length or content is invalid for the frame kind.
+ public static EventFrame DecodeBody(ushort kind, ReadOnlySpan body)
+ {
+ switch (kind)
+ {
+ case (ushort)EventFrameKind.Hello:
+ ExpectLength(kind, body, 4);
+ return new Hello(
+ BinaryPrimitives.ReadUInt16LittleEndian(body),
+ BinaryPrimitives.ReadUInt16LittleEndian(body[2..]));
+
+ case (ushort)EventFrameKind.StatusUpdated:
+ ExpectLength(kind, body, 0);
+ return new StatusUpdated();
+
+ case (ushort)EventFrameKind.Stdout:
+ return new Stdout(DecodeUtf8(body));
+
+ case (ushort)EventFrameKind.Stderr:
+ return new Stderr(DecodeUtf8(body));
+
+ case (ushort)EventFrameKind.Finish:
+ ExpectLength(kind, body, 0);
+ return new Finish();
+
+ case (ushort)EventFrameKind.StdoutOverflow:
+ ExpectLength(kind, body, 4);
+ return new StdoutOverflow(BinaryPrimitives.ReadUInt32LittleEndian(body));
+
+ case (ushort)EventFrameKind.StderrOverflow:
+ ExpectLength(kind, body, 4);
+ return new StderrOverflow(BinaryPrimitives.ReadUInt32LittleEndian(body));
+
+ default:
+ return new Unknown(kind, body.ToArray());
+ }
+ }
+
+ private static byte[] EncodeHelloBody(Hello hello)
+ {
+ var body = new byte[4];
+ BinaryPrimitives.WriteUInt16LittleEndian(body, hello.VersionMajor);
+ BinaryPrimitives.WriteUInt16LittleEndian(body.AsSpan(2), hello.VersionMinor);
+ return body;
+ }
+
+ private static byte[] EncodeOverflowBody(uint bytesSkipped)
+ {
+ var body = new byte[4];
+ BinaryPrimitives.WriteUInt32LittleEndian(body, bytesSkipped);
+ return body;
+ }
+
+ private static void ExpectLength(ushort kind, ReadOnlySpan body, int expected)
+ {
+ if (body.Length != expected)
+ {
+ throw new EventFrameException(
+ $"Frame kind 0x{kind:x4} expects a body of {expected} bytes, got {body.Length}.");
+ }
+ }
+
+ private static string DecodeUtf8(ReadOnlySpan body)
+ {
+ try
+ {
+ return new UTF8Encoding(encoderShouldEmitUTF8Identifier: false, throwOnInvalidBytes: true).GetString(body);
+ }
+ catch (DecoderFallbackException e)
+ {
+ throw new EventFrameException("Stdout/stderr frame body is not valid UTF-8.", e);
+ }
+ }
+}
+
+/// Error produced while encoding or decoding event frames.
+public sealed class EventFrameException : Exception
+{
+ public EventFrameException(string message)
+ : base(message)
+ {
+ }
+
+ public EventFrameException(string message, Exception innerException)
+ : base(message, innerException)
+ {
+ }
+}
+
+///
+/// Incremental decoder for a NOW_BROKER event frame stream. Feed raw bytes read
+/// from the channel with , then drain complete frames with
+/// .
+///
+public sealed class EventFrameDecoder
+{
+ private readonly MemoryStream _buffer = new();
+
+ ///
+ /// True when the decoder holds buffered bytes that do not yet form a complete frame.
+ /// If the transport reaches end-of-stream while this is true, the frame stream was
+ /// truncated mid-frame.
+ ///
+ public bool HasBufferedData => _buffer.Length > 0;
+
+ /// Append raw bytes received from the transport.
+ public void Extend(ReadOnlySpan bytes)
+ {
+ _buffer.Write(bytes);
+ }
+
+ ///
+ /// Try to decode the next complete frame. Returns false when more bytes are needed.
+ /// Frames with unknown kinds are returned as and
+ /// should be ignored by the consumer.
+ ///
+ ///
+ /// The stream is corrupt (oversized frame, malformed body, or invalid UTF-8); the
+ /// channel should be closed.
+ ///
+ public bool TryReadFrame(out EventFrame? frame)
+ {
+ frame = null;
+ var buffered = _buffer.GetBuffer().AsSpan(0, (int)_buffer.Length);
+ if (buffered.Length < EventChannelProtocol.FrameHeaderSize)
+ {
+ return false;
+ }
+
+ var size = BinaryPrimitives.ReadUInt32LittleEndian(buffered);
+ if (size > EventChannelProtocol.MaxFrameBodyBytes)
+ {
+ throw new EventFrameException(
+ $"Frame body size {size} exceeds the maximum of {EventChannelProtocol.MaxFrameBodyBytes} bytes.");
+ }
+
+ var total = EventChannelProtocol.FrameHeaderSize + (int)size;
+ if (buffered.Length < total)
+ {
+ return false;
+ }
+
+ var kind = BinaryPrimitives.ReadUInt16LittleEndian(buffered[4..]);
+ frame = EventFrame.DecodeBody(kind, buffered[EventChannelProtocol.FrameHeaderSize..total]);
+
+ var remainder = buffered[total..].ToArray();
+ _buffer.SetLength(0);
+ _buffer.Write(remainder);
+ return true;
+ }
+}
\ No newline at end of file
diff --git a/policies/dotnet/Devolutions.Now.Policy.Api/RequestModels.cs b/policies/dotnet/Devolutions.Now.Policy.Api/RequestModels.cs
index eea5fc6..f2dd5bb 100644
--- a/policies/dotnet/Devolutions.Now.Policy.Api/RequestModels.cs
+++ b/policies/dotnet/Devolutions.Now.Policy.Api/RequestModels.cs
@@ -51,8 +51,11 @@ public string RequestKind
public bool IncludeCommandPreview { get; set; }
///
- /// When true, the broker captures the operation's combined stdout+stderr and returns it
- /// (tail-truncated) in the status response. Off by default to avoid the overhead when not needed.
+ /// When true, the operation's stdout/stderr data is pushed over the per-operation
+ /// event channel (see ). The channel
+ /// itself is opened unconditionally when supported and always carries status change
+ /// notifications; this flag only controls whether output data frames are sent. Off
+ /// by default to avoid the overhead when the client does not need the output.
///
[JsonPropertyName("CaptureOutput")]
public bool CaptureOutput { get; set; }
diff --git a/policies/dotnet/Devolutions.Now.Policy.Api/ResponseModels.cs b/policies/dotnet/Devolutions.Now.Policy.Api/ResponseModels.cs
index d686782..d86dd57 100644
--- a/policies/dotnet/Devolutions.Now.Policy.Api/ResponseModels.cs
+++ b/policies/dotnet/Devolutions.Now.Policy.Api/ResponseModels.cs
@@ -157,4 +157,13 @@ public sealed class OperationSubmission
[JsonPropertyName("SubmittedAt")]
public DateTimeOffset SubmittedAt { get; set; }
+
+ ///
+ /// Per-operation event channel carrying NOW_BROKER event frames (status change
+ /// notifications and, when the execute request opted in via
+ /// , stdout/stderr data). Present whenever
+ /// the broker supports event channels; absent otherwise.
+ ///
+ [JsonPropertyName("EventChannel")]
+ public EventChannel? EventChannel { get; set; }
}
\ No newline at end of file
diff --git a/policies/dotnet/Devolutions.Now.Policy.Api/StatusModels.cs b/policies/dotnet/Devolutions.Now.Policy.Api/StatusModels.cs
index fbf6249..1bb72e3 100644
--- a/policies/dotnet/Devolutions.Now.Policy.Api/StatusModels.cs
+++ b/policies/dotnet/Devolutions.Now.Policy.Api/StatusModels.cs
@@ -70,11 +70,4 @@ public string ResponseKind
[JsonPropertyName("Details")]
public JsonNode? Details { get; set; }
-
- ///
- /// Captured combined stdout+stderr of the operation as base64-encoded UTF-8 data (tail-truncated to ~10 KiB before encoding).
- /// Only present when the request opted in via .
- ///
- [JsonPropertyName("Stdout")]
- public string? Stdout { get; set; }
}
\ No newline at end of file
diff --git a/policies/dotnet/Devolutions.Now.Policy.Client.Tests/EventChannelTests.cs b/policies/dotnet/Devolutions.Now.Policy.Client.Tests/EventChannelTests.cs
new file mode 100644
index 0000000..da327c5
--- /dev/null
+++ b/policies/dotnet/Devolutions.Now.Policy.Client.Tests/EventChannelTests.cs
@@ -0,0 +1,114 @@
+using Xunit;
+
+namespace Devolutions.Now.Policy.Client.Tests;
+
+///
+/// Tests for the NOW_BROKER event channel frame protocol. The binary fixture is
+/// shared with the Rust test suite so both implementations stay wire-compatible.
+///
+public class EventChannelTests
+{
+ private static string FramesFixturePath =>
+ Path.Combine(TestData.SamplesDir, "frames", "event-channel.frames.bin");
+
+ [Fact]
+ public void Shared_frame_fixture_decodes_to_expected_frames()
+ {
+ var bytes = File.ReadAllBytes(FramesFixturePath);
+
+ var frames = DecodeAll(bytes);
+
+ Assert.Collection(
+ frames,
+ f =>
+ {
+ var hello = Assert.IsType(f);
+ Assert.Equal(EventChannelProtocol.VersionMajor, hello.VersionMajor);
+ Assert.Equal(EventChannelProtocol.VersionMinor, hello.VersionMinor);
+ },
+ f => Assert.IsType(f),
+ f => Assert.Equal("hello \u03c0\n", Assert.IsType(f).Data),
+ f => Assert.Equal("oops\n", Assert.IsType(f).Data),
+ f => Assert.Equal(4096u, Assert.IsType(f).BytesSkipped),
+ f => Assert.Equal(16u, Assert.IsType(f).BytesSkipped),
+ f =>
+ {
+ var unknown = Assert.IsType(f);
+ Assert.Equal(0x7fff, unknown.Kind);
+ Assert.Equal([1, 2, 3], unknown.Body);
+ },
+ f => Assert.IsType(f));
+
+ // The fixture must round-trip byte-for-byte through the encoder.
+ var reencoded = frames.SelectMany(f => f.Encode()).ToArray();
+ Assert.Equal(bytes, reencoded);
+ }
+
+ [Fact]
+ public void Decoder_handles_partial_input_byte_by_byte()
+ {
+ var bytes = File.ReadAllBytes(FramesFixturePath);
+
+ var decoder = new EventFrameDecoder();
+ var frames = new List();
+ foreach (var b in bytes)
+ {
+ decoder.Extend([b]);
+ while (decoder.TryReadFrame(out var frame))
+ {
+ frames.Add(frame!);
+ }
+ }
+
+ Assert.Equal(8, frames.Count);
+ Assert.IsType(frames[0]);
+ Assert.IsType(frames[^1]);
+ }
+
+ [Fact]
+ public void Decoder_rejects_invalid_utf8_output_frames()
+ {
+ var decoder = new EventFrameDecoder();
+ decoder.Extend([2, 0, 0, 0, 0x03, 0x00, 0xff, 0xfe]);
+
+ Assert.Throws(() => decoder.TryReadFrame(out _));
+ }
+
+ [Fact]
+ public void Decoder_rejects_oversized_frames()
+ {
+ var decoder = new EventFrameDecoder();
+ var size = BitConverter.GetBytes((uint)EventChannelProtocol.MaxFrameBodyBytes + 1);
+ decoder.Extend([size[0], size[1], size[2], size[3], 0x03, 0x00]);
+
+ Assert.Throws(() => decoder.TryReadFrame(out _));
+ }
+
+ [Fact]
+ public void Fixed_body_frames_reject_wrong_length()
+ {
+ Assert.Throws(() => EventFrame.DecodeBody((ushort)EventFrameKind.Hello, [1, 0]));
+ Assert.Throws(() => EventFrame.DecodeBody((ushort)EventFrameKind.Finish, [0]));
+ }
+
+ [Fact]
+ public void Encoder_rejects_oversized_stdout_data()
+ {
+ var frame = new EventFrame.Stdout(new string('a', EventChannelProtocol.MaxFrameBodyBytes + 1));
+
+ Assert.Throws(() => frame.Encode());
+ }
+
+ private static List DecodeAll(byte[] bytes)
+ {
+ var decoder = new EventFrameDecoder();
+ decoder.Extend(bytes);
+ var frames = new List();
+ while (decoder.TryReadFrame(out var frame))
+ {
+ frames.Add(frame!);
+ }
+
+ return frames;
+ }
+}
\ No newline at end of file
diff --git a/policies/dotnet/Devolutions.Now.Policy.Client.Tests/OperationEventChannelTests.cs b/policies/dotnet/Devolutions.Now.Policy.Client.Tests/OperationEventChannelTests.cs
new file mode 100644
index 0000000..99f7e27
--- /dev/null
+++ b/policies/dotnet/Devolutions.Now.Policy.Client.Tests/OperationEventChannelTests.cs
@@ -0,0 +1,223 @@
+using System.IO.Pipes;
+
+using Xunit;
+
+namespace Devolutions.Now.Policy.Client.Tests;
+
+///
+/// Tests for and
+/// using a real local named pipe carrying the shared
+/// binary frame fixture.
+///
+public class OperationEventChannelTests
+{
+ private static string FramesFixturePath =>
+ Path.Combine(TestData.SamplesDir, "frames", "event-channel.frames.bin");
+
+ [Fact]
+ public async Task OpenEventChannel_reads_events_pushed_over_local_pipe()
+ {
+ var pipeName = RandomPipeName();
+ using var server = ServeFixtureBytes(pipeName, File.ReadAllBytes(FramesFixturePath));
+ using var client = CreateClient();
+
+ await using var channel = await client.OpenEventChannel(CreateExecutionResponse(pipeName));
+
+ var frames = new List();
+ await foreach (var frame in channel.ReadEvents())
+ {
+ frames.Add(frame);
+ }
+
+ // ReadEvents skips the Unknown frame in the fixture and completes after Finish.
+ Assert.Collection(
+ frames,
+ f =>
+ {
+ var hello = Assert.IsType(f);
+ Assert.Equal(EventChannelProtocol.VersionMajor, hello.VersionMajor);
+ Assert.Equal(EventChannelProtocol.VersionMinor, hello.VersionMinor);
+ },
+ f => Assert.IsType(f),
+ f => Assert.Equal("hello \u03c0\n", Assert.IsType(f).Data),
+ f => Assert.Equal("oops\n", Assert.IsType(f).Data),
+ f => Assert.Equal(4096u, Assert.IsType(f).BytesSkipped),
+ f => Assert.Equal(16u, Assert.IsType(f).BytesSkipped),
+ f => Assert.IsType(f));
+ }
+
+ [Fact]
+ public async Task ReadFrame_returns_unknown_frames_and_null_at_end_of_stream()
+ {
+ var pipeName = RandomPipeName();
+ using var server = ServeFixtureBytes(pipeName, File.ReadAllBytes(FramesFixturePath));
+ using var client = CreateClient();
+
+ await using var channel = await client.OpenEventChannel(CreateExecutionResponse(pipeName));
+
+ var frames = new List();
+ while (await channel.ReadFrame() is { } frame)
+ {
+ frames.Add(frame);
+ }
+
+ Assert.Equal(8, frames.Count);
+ Assert.Equal(0x7fff, Assert.IsType(frames[6]).Kind);
+ Assert.IsType(frames[^1]);
+ }
+
+ [Fact]
+ public async Task OpenEventChannel_rejects_response_without_event_channel()
+ {
+ using var client = CreateClient();
+ var response = CreateExecutionResponse(pipeName: null);
+
+ var ex = await Assert.ThrowsAsync(() => client.OpenEventChannel(response));
+ Assert.Equal(BrokerClientErrorKind.InvalidResponse, ex.Kind);
+ }
+
+ [Fact]
+ public async Task OpenEventChannel_rejects_empty_channel_path()
+ {
+ using var client = CreateClient();
+ var response = CreateExecutionResponse(pipeName: "");
+
+ var ex = await Assert.ThrowsAsync(() => client.OpenEventChannel(response));
+ Assert.Equal(BrokerClientErrorKind.InvalidResponse, ex.Kind);
+ }
+
+ [Fact]
+ public async Task ReadFrame_returns_null_when_server_closes_pipe_at_frame_boundary()
+ {
+ var pipeName = RandomPipeName();
+ // Only Hello and one stdout frame; the server closes the pipe without sending Finish.
+ var bytes = new EventFrame.Hello(1, 0).Encode()
+ .Concat(new EventFrame.Stdout("partial run\n").Encode())
+ .ToArray();
+ using var server = ServeFixtureBytes(pipeName, bytes);
+ using var client = CreateClient();
+
+ await using var channel = await client.OpenEventChannel(CreateExecutionResponse(pipeName));
+
+ Assert.IsType(await channel.ReadFrame());
+ Assert.IsType(await channel.ReadFrame());
+ Assert.Null(await channel.ReadFrame());
+ }
+
+ [Fact]
+ public async Task ReadEvents_completes_without_finish_when_server_closes_pipe()
+ {
+ var pipeName = RandomPipeName();
+ var bytes = new EventFrame.Hello(1, 0).Encode()
+ .Concat(new EventFrame.StatusUpdated().Encode())
+ .ToArray();
+ using var server = ServeFixtureBytes(pipeName, bytes);
+ using var client = CreateClient();
+
+ await using var channel = await client.OpenEventChannel(CreateExecutionResponse(pipeName));
+
+ var frames = new List();
+ await foreach (var frame in channel.ReadEvents())
+ {
+ frames.Add(frame);
+ }
+
+ Assert.Equal(2, frames.Count);
+ Assert.DoesNotContain(frames, f => f is EventFrame.Finish);
+ }
+
+ [Fact]
+ public async Task ReadFrame_throws_when_server_closes_pipe_mid_frame()
+ {
+ var pipeName = RandomPipeName();
+ // A complete Hello followed by a truncated stdout frame: the header announces
+ // an 11-byte body but the server closes the pipe after 4 body bytes.
+ var truncated = new EventFrame.Stdout("interrupted").Encode()[..10];
+ var bytes = new EventFrame.Hello(1, 0).Encode().Concat(truncated).ToArray();
+ using var server = ServeFixtureBytes(pipeName, bytes);
+ using var client = CreateClient();
+
+ await using var channel = await client.OpenEventChannel(CreateExecutionResponse(pipeName));
+
+ Assert.IsType(await channel.ReadFrame());
+ await Assert.ThrowsAsync(() => channel.ReadFrame());
+ }
+
+ [Fact]
+ public async Task ReadFrame_rejects_stream_that_does_not_start_with_hello()
+ {
+ var pipeName = RandomPipeName();
+ var bytes = new EventFrame.Stdout("data before hello\n").Encode();
+ using var server = ServeFixtureBytes(pipeName, bytes);
+ using var client = CreateClient();
+
+ await using var channel = await client.OpenEventChannel(CreateExecutionResponse(pipeName));
+
+ var ex = await Assert.ThrowsAsync(() => channel.ReadFrame());
+ Assert.Contains("Hello", ex.Message);
+ }
+
+ [Fact]
+ public async Task ReadFrame_rejects_unsupported_protocol_major_version()
+ {
+ var pipeName = RandomPipeName();
+ var bytes = new EventFrame.Hello((ushort)(EventChannelProtocol.VersionMajor + 1), 0).Encode();
+ using var server = ServeFixtureBytes(pipeName, bytes);
+ using var client = CreateClient();
+
+ await using var channel = await client.OpenEventChannel(CreateExecutionResponse(pipeName));
+
+ var ex = await Assert.ThrowsAsync(() => channel.ReadFrame());
+ Assert.Contains("major version", ex.Message);
+ }
+
+ // Keep the name short: on macOS named pipes map to unix domain socket paths
+ // (temp dir + "CoreFxPipe_" prefix) limited to 104 characters.
+ private static string RandomPipeName() =>
+ $"nowec-{Guid.NewGuid():N}"[..22];
+
+ private static ExecutionResponse CreateExecutionResponse(string? pipeName) => new()
+ {
+ Operation = new OperationSubmission
+ {
+ OperationId = "op-test-000001",
+ EventChannel = pipeName is null
+ ? null
+ : new EventChannel { Kind = EventChannelKind.LocalPipe, Path = pipeName },
+ },
+ };
+
+ private static BrokerClient CreateClient() => new(new BrokerClientOptions
+ {
+ EffectiveUser = "DEVOLUTIONS\\bob",
+ RequestedElevation = Elevation.Standard,
+ ClientExecutablePath = "C:\\Tools\\client.exe",
+ ClientVersion = "9.8.7",
+ });
+
+ ///
+ /// Start a one-shot pipe server that writes to the first
+ /// connected client and then closes the pipe.
+ ///
+ private static IDisposable ServeFixtureBytes(string pipeName, byte[] bytes)
+ {
+ var server = new NamedPipeServerStream(
+ pipeName,
+ PipeDirection.Out,
+ maxNumberOfServerInstances: 1,
+ PipeTransmissionMode.Byte,
+ PipeOptions.Asynchronous);
+
+ _ = Task.Run(async () =>
+ {
+ await using (server)
+ {
+ await server.WaitForConnectionAsync();
+ await server.WriteAsync(bytes);
+ await server.FlushAsync();
+ }
+ });
+
+ return server;
+ }
+}
\ No newline at end of file
diff --git a/policies/dotnet/Devolutions.Now.Policy.Client/BrokerClient.cs b/policies/dotnet/Devolutions.Now.Policy.Client/BrokerClient.cs
index ad53bdf..1ee52bd 100644
--- a/policies/dotnet/Devolutions.Now.Policy.Client/BrokerClient.cs
+++ b/policies/dotnet/Devolutions.Now.Policy.Client/BrokerClient.cs
@@ -245,6 +245,45 @@ public async Task QueryStatus(
return DeserializeResponse(response, "status", "/v1/package-operations/get-status");
}
+ ///
+ /// Connect to the per-operation event channel advertised in an execution response and
+ /// return a reader for NOW_BROKER event frames (stdout/stderr data and status
+ /// change notifications).
+ ///
+ ///
+ /// The response carries no event channel descriptor, the channel kind is not
+ /// supported, or the connection failed.
+ ///
+ public async Task OpenEventChannel(
+ ExecutionResponse response,
+ CancellationToken cancellationToken = default)
+ {
+ ArgumentNullException.ThrowIfNull(response);
+
+ var descriptor = response.Operation?.EventChannel
+ ?? throw new BrokerClientException(
+ BrokerClientErrorKind.InvalidResponse,
+ "The execution response does not carry an event channel descriptor; "
+ + "the broker likely does not support event channels.");
+
+ if (descriptor.Kind != EventChannelKind.LocalPipe)
+ {
+ throw new BrokerClientException(
+ BrokerClientErrorKind.UnsupportedCapability,
+ $"Unsupported event channel kind '{descriptor.Kind}'.");
+ }
+
+ if (string.IsNullOrWhiteSpace(descriptor.Path))
+ {
+ throw new BrokerClientException(
+ BrokerClientErrorKind.InvalidResponse,
+ "The event channel descriptor carries an empty path.");
+ }
+
+ Trace?.Invoke($"Connecting to operation event channel pipe '{descriptor.Path}'.");
+ return await OperationEventChannel.ConnectLocalPipe(descriptor.Path, cancellationToken).ConfigureAwait(false);
+ }
+
public void Dispose()
{
_transport.Dispose();
diff --git a/policies/dotnet/Devolutions.Now.Policy.Client/OperationEventChannel.cs b/policies/dotnet/Devolutions.Now.Policy.Client/OperationEventChannel.cs
new file mode 100644
index 0000000..20b3d02
--- /dev/null
+++ b/policies/dotnet/Devolutions.Now.Policy.Client/OperationEventChannel.cs
@@ -0,0 +1,177 @@
+using System.IO.Pipes;
+using System.Runtime.CompilerServices;
+
+using Devolutions.Now.Policy.Api;
+
+namespace Devolutions.Now.Policy.Client;
+
+///
+/// Client side of a per-operation event channel carrying the NOW_BROKER frame
+/// protocol. Obtain an instance with
+/// .
+///
+///
+/// The channel is one-way: the broker pushes stdout/stderr data and status change
+/// notifications, the client only reads. The first frame is always
+/// ; after the channel can
+/// be disposed. See policies/docs/event-channel-protocol.md.
+///
+public sealed class OperationEventChannel : IAsyncDisposable, IDisposable
+{
+ private const int ConnectTimeoutMs = 5000;
+ private const int ReadBufferSize = 8192;
+
+ private readonly Stream _stream;
+ private readonly EventFrameDecoder _decoder = new();
+ private readonly byte[] _readBuffer = new byte[ReadBufferSize];
+ private bool _helloReceived;
+
+ internal OperationEventChannel(Stream stream)
+ {
+ _stream = stream;
+ }
+
+ internal static async Task ConnectLocalPipe(string path, CancellationToken cancellationToken)
+ {
+ var pipeName = NormalizePipeName(path);
+ var pipe = new NamedPipeClientStream(".", pipeName, PipeDirection.In, PipeOptions.Asynchronous);
+
+ try
+ {
+ using var connectCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+ connectCts.CancelAfter(ConnectTimeoutMs);
+ await pipe.ConnectAsync(connectCts.Token).ConfigureAwait(false);
+ return new OperationEventChannel(pipe);
+ }
+ catch (OperationCanceledException ex) when (!cancellationToken.IsCancellationRequested)
+ {
+ await pipe.DisposeAsync().ConfigureAwait(false);
+ throw new BrokerClientException(
+ BrokerClientErrorKind.Timeout,
+ $"Timed out connecting to the operation event channel pipe '{pipeName}'.",
+ innerException: ex);
+ }
+ catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
+ {
+ await pipe.DisposeAsync().ConfigureAwait(false);
+ throw new BrokerClientException(
+ BrokerClientErrorKind.BrokerUnavailable,
+ $"Unable to connect to the operation event channel pipe '{pipeName}': {ex.Message}",
+ innerException: ex);
+ }
+ catch
+ {
+ await pipe.DisposeAsync().ConfigureAwait(false);
+ throw;
+ }
+ }
+
+ ///
+ /// Read the next frame from the channel. Returns null when the broker closed
+ /// the channel. Frames with unknown kinds are returned as
+ /// and should be ignored by the caller.
+ ///
+ ///
+ /// The frame stream is corrupt, the channel was closed mid-frame, the stream does
+ /// not start with a Hello frame, or the advertised protocol major version is
+ /// unsupported; the channel should be disposed.
+ ///
+ /// The transport failed.
+ public async Task ReadFrame(CancellationToken cancellationToken = default)
+ {
+ while (true)
+ {
+ if (_decoder.TryReadFrame(out var frame))
+ {
+ EnforceHandshake(frame!);
+ return frame;
+ }
+
+ var bytesRead = await _stream.ReadAsync(_readBuffer, cancellationToken).ConfigureAwait(false);
+ if (bytesRead == 0)
+ {
+ if (_decoder.HasBufferedData)
+ {
+ throw new EventFrameException("The event channel was closed mid-frame; the frame stream is truncated.");
+ }
+
+ return null;
+ }
+
+ _decoder.Extend(_readBuffer.AsSpan(0, bytesRead));
+ }
+ }
+
+ private void EnforceHandshake(EventFrame frame)
+ {
+ if (_helloReceived)
+ {
+ return;
+ }
+
+ if (frame is not EventFrame.Hello hello)
+ {
+ throw new EventFrameException(
+ $"The event channel did not start with a Hello frame (got kind 0x{frame.Kind:x4}).");
+ }
+
+ if (hello.VersionMajor != EventChannelProtocol.VersionMajor)
+ {
+ throw new EventFrameException(
+ $"Unsupported event channel protocol major version {hello.VersionMajor}; "
+ + $"this client supports version {EventChannelProtocol.VersionMajor}.");
+ }
+
+ _helloReceived = true;
+ }
+
+ ///
+ /// Asynchronously enumerate operation events. Frames with unknown kinds are skipped;
+ /// the sequence completes after or when the broker
+ /// closes the channel.
+ ///
+ /// The frame stream is corrupt; the channel should be disposed.
+ /// The transport failed.
+ public async IAsyncEnumerable ReadEvents([EnumeratorCancellation] CancellationToken cancellationToken = default)
+ {
+ while (true)
+ {
+ var frame = await ReadFrame(cancellationToken).ConfigureAwait(false);
+ if (frame is null)
+ {
+ yield break;
+ }
+
+ if (frame is EventFrame.Unknown)
+ {
+ continue;
+ }
+
+ yield return frame;
+
+ if (frame is EventFrame.Finish)
+ {
+ yield break;
+ }
+ }
+ }
+
+ public void Dispose()
+ {
+ _stream.Dispose();
+ }
+
+ public ValueTask DisposeAsync()
+ {
+ return _stream.DisposeAsync();
+ }
+
+ private static string NormalizePipeName(string path)
+ {
+ const string win32PipePrefix = @"\\.\pipe\";
+
+ return path.StartsWith(win32PipePrefix, StringComparison.OrdinalIgnoreCase)
+ ? path[win32PipePrefix.Length..]
+ : path;
+ }
+}
\ No newline at end of file
diff --git a/policies/dotnet/Devolutions.Now.Policy.Client/README.md b/policies/dotnet/Devolutions.Now.Policy.Client/README.md
index 2a3e945..043f6f3 100644
--- a/policies/dotnet/Devolutions.Now.Policy.Client/README.md
+++ b/policies/dotnet/Devolutions.Now.Policy.Client/README.md
@@ -16,6 +16,28 @@ The client is used to:
- submit package operations for elevated execution;
- poll asynchronous operation status until completion or failure.
+Execution responses additionally return a per-operation event channel descriptor (`OperationSubmission.EventChannel`) whenever the broker supports event channels. The channel carries the `NOW_BROKER` frame protocol: status change notifications pushed by the broker and, when the operation was submitted with `CaptureOutput`, stdout/stderr data. The frame codec (`EventFrame`, `EventFrameDecoder`) lives in `Devolutions.Now.Policy.Api`; see `policies/docs/event-channel-protocol.md` for the wire specification.
+
+To consume the channel, pass the execution response to `BrokerClient.OpenEventChannel`; it connects to the advertised local pipe and returns an `OperationEventChannel`:
+
+```csharp
+var execution = await client.Execute(request);
+await using var channel = await client.OpenEventChannel(execution);
+
+await foreach (var frame in channel.ReadEvents())
+{
+ switch (frame)
+ {
+ case EventFrame.Stdout stdout: Console.Out.Write(stdout.Data); break;
+ case EventFrame.Stderr stderr: Console.Error.Write(stderr.Data); break;
+ case EventFrame.StatusUpdated: /* issue QueryStatus */ break;
+ case EventFrame.Finish: /* operation finished; enumeration completes */ break;
+ }
+}
+```
+
+`ReadEvents` skips unknown frame kinds and completes after `Finish` or when the broker closes the channel; `ReadFrame` exposes the raw frame stream, including `EventFrame.Unknown`.
+
Architecture
------------
@@ -28,6 +50,7 @@ The main surface is `BrokerClient`:
- `ExecuteAndWait` submits an operation and polls status until a terminal state.
- `QueryStatus` sends `POST /v1/package-operations/get-status`.
- `Cancel` sends `POST /v1/package-operations/cancel` to request cancelation of an in-flight operation.
+- `OpenEventChannel` connects to the per-operation event channel advertised in an `ExecutionResponse` and returns an `OperationEventChannel` frame reader.
Transport is abstracted behind `IBrokerTransport`, which exchanges HTTP-style `BrokerTransportRequest` and `BrokerTransportResponse` values. `NamedPipeBrokerTransport` is the default implementation and sends HTTP/1.1 over a Windows named pipe. Tests and future transports can inject their own transport through `BrokerClientOptions.Transport`.
diff --git a/policies/rust/now-policy-api/Cargo.toml b/policies/rust/now-policy-api/Cargo.toml
index f32d820..5943701 100644
--- a/policies/rust/now-policy-api/Cargo.toml
+++ b/policies/rust/now-policy-api/Cargo.toml
@@ -18,7 +18,6 @@ default = []
policy-compat = ["dep:now-policy"]
[dependencies]
-base64 = "0.22"
chrono = { version = "0.4", features = ["serde"] }
derive_more = { version = "2", features = ["as_ref", "deref", "display", "from"] }
now-policy = { version = "0.2", path = "../now-policy", optional = true }
diff --git a/policies/rust/now-policy-api/README.md b/policies/rust/now-policy-api/README.md
index 9e5b4d5..7a20c35 100644
--- a/policies/rust/now-policy-api/README.md
+++ b/policies/rust/now-policy-api/README.md
@@ -24,6 +24,7 @@ Library structure overview:
- `execute.rs` contains execution response models for `POST /v1/package-operations/execute`.
- `evaluate.rs` contains evaluation response models for `POST /v1/package-operations/evaluate`.
- `status.rs` contains status request/response models for `POST /v1/package-operations/get-status`.
+- `event_channel.rs` contains the per-operation event channel descriptor returned in execution responses and the `NOW_BROKER` binary frame protocol codec (see `policies/docs/event-channel-protocol.md`).
- `health.rs` contains health endpoint models for `GET /v1/health`.
- `capabilities.rs` contains capability endpoint models for `GET /v1/capabilities`.
- `enums.rs` contains shared protocol enums.
diff --git a/policies/rust/now-policy-api/openapi/now-policy-api.yaml b/policies/rust/now-policy-api/openapi/now-policy-api.yaml
index 2adf68d..80eff14 100644
--- a/policies/rust/now-policy-api/openapi/now-policy-api.yaml
+++ b/policies/rust/now-policy-api/openapi/now-policy-api.yaml
@@ -181,11 +181,6 @@ components:
- X64
- Arm64
- Neutral
- Base64Utf8Data:
- description: Base64-encoded UTF-8 operation output.
- type: string
- maxLength: 16384
- pattern: ^[A-Za-z0-9+/]*={0,2}$
CancelRequest:
description: Request body for canceling a previously submitted operation.
type: object
@@ -495,6 +490,29 @@ components:
EvaluationResponseKind:
type: string
pattern: ^EvaluationResponse$
+ EventChannel:
+ description: Descriptor of a per-operation event channel returned in the execution response.
+ type: object
+ required:
+ - Kind
+ - Path
+ properties:
+ Kind:
+ description: Transport kind of the channel.
+ $ref: '#/components/schemas/EventChannelKind'
+ Path:
+ description: Transport-specific path. For the `LocalPipe` kind this is the pipe name/path the client should connect to.
+ type: string
+ maxLength: 1024
+ minLength: 1
+ additionalProperties: false
+ EventChannelKind:
+ description: Transport kind of a per-operation event channel.
+ oneOf:
+ - description: Local named pipe carrying `NOW_BROKER` event frames.
+ type: string
+ enum:
+ - LocalPipe
ExecutionResponse:
description: Response returned after an execute request is evaluated and, when allowed, submitted.
type: object
@@ -712,6 +730,10 @@ components:
- Status
- SubmittedAt
properties:
+ EventChannel:
+ description: Per-operation event channel carrying `NOW_BROKER` event frames (status change notifications and, when the execute request opted in via `CaptureOutput`, stdout/stderr data). Present whenever the broker supports event channels; absent otherwise.
+ $ref: '#/components/schemas/EventChannel'
+ nullable: true
OperationId:
description: Server-issued stable operation identifier.
$ref: '#/components/schemas/ResourceId'
@@ -745,7 +767,7 @@ components:
- Source
properties:
CaptureOutput:
- description: When true, the broker captures the operation's combined stdout+stderr and returns it (tail-truncated) in the status response. Off by default to avoid the overhead when the client does not need the output.
+ description: When true, the operation's stdout/stderr data is pushed over the per-operation event channel (see the `EventChannel` descriptor in the execution response). The channel itself is opened unconditionally when supported and always carries status change notifications; this flag only controls whether output data frames are sent. Off by default to avoid the overhead when the client does not need the output.
default: false
type: boolean
Client:
@@ -1054,10 +1076,6 @@ components:
Status:
description: Current status of the operation.
$ref: '#/components/schemas/OperationStatus'
- Stdout:
- description: Captured combined stdout+stderr as base64-encoded UTF-8 data (tail-truncated to ~10 KiB before encoding). Only present when the original request opted in via `CaptureOutput`.
- $ref: '#/components/schemas/Base64Utf8Data'
- nullable: true
additionalProperties: false
StatusResponseKind:
type: string
diff --git a/policies/rust/now-policy-api/src/api.rs b/policies/rust/now-policy-api/src/api.rs
index 31c5536..00ec0b4 100644
--- a/policies/rust/now-policy-api/src/api.rs
+++ b/policies/rust/now-policy-api/src/api.rs
@@ -55,9 +55,12 @@ pub struct PackageRequest {
#[serde(default)]
pub include_command_preview: bool,
- /// When true, the broker captures the operation's combined stdout+stderr and returns
- /// it (tail-truncated) in the status response. Off by default to avoid the overhead
- /// when the client does not need the output.
+ /// When true, the operation's stdout/stderr data is pushed over the
+ /// per-operation event channel (see the `EventChannel` descriptor in the
+ /// execution response). The channel itself is opened unconditionally when
+ /// supported and always carries status change notifications; this flag only
+ /// controls whether output data frames are sent. Off by default to avoid the
+ /// overhead when the client does not need the output.
#[serde(default)]
pub capture_output: bool,
}
diff --git a/policies/rust/now-policy-api/src/enums.rs b/policies/rust/now-policy-api/src/enums.rs
index 3ad8074..29f0dcb 100644
--- a/policies/rust/now-policy-api/src/enums.rs
+++ b/policies/rust/now-policy-api/src/enums.rs
@@ -102,6 +102,14 @@ impl OperationStatus {
}
}
+/// Transport kind of a per-operation event channel.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema, strum::Display)]
+#[schemars(rename = "EventChannelKind")]
+pub enum EventChannelKind {
+ /// Local named pipe carrying `NOW_BROKER` event frames.
+ LocalPipe,
+}
+
/// Structured machine-readable error code.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[schemars(rename = "ErrorCode")]
diff --git a/policies/rust/now-policy-api/src/event_channel.rs b/policies/rust/now-policy-api/src/event_channel.rs
new file mode 100644
index 0000000..10703f2
--- /dev/null
+++ b/policies/rust/now-policy-api/src/event_channel.rs
@@ -0,0 +1,420 @@
+//! Per-operation event channel descriptor and `NOW_BROKER` frame protocol.
+//!
+//! For each executed operation the broker opens a per-operation event channel
+//! (currently a local named pipe), when supported, and returns its descriptor
+//! in the execution response. The channel is one-way (read-only from the
+//! client side) and carries a minimal length-prefixed binary frame protocol
+//! used to:
+//!
+//! - push both stdout and stderr data over a single channel (only when the
+//! execute request opts in via `CaptureOutput`), preserving the sequential
+//! order of the interleaved output;
+//! - notify the client that the operation status changed, so status is only
+//! queried over HTTP when something actually happened (no periodic polling);
+//! - signal operation completion, after which the channel can be closed.
+//!
+//! # Wire format
+//!
+//! All integers are little-endian. Each frame is:
+//!
+//! ```text
+//! | u32 frame_size | u16 frame_kind | [frame_size; u8] frame_body |
+//! ```
+//!
+//! `frame_size` is the body length in bytes and excludes the 6-byte header.
+//! Stdout/stderr frame bodies are strictly UTF-8; the broker never splits a
+//! UTF-8 character across frames. Decoders must skip frames with unknown
+//! kinds so new frame kinds can be added without breaking older clients.
+
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+use super::enums::EventChannelKind;
+
+/// Descriptor of a per-operation event channel returned in the execution response.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
+#[schemars(rename = "EventChannel")]
+#[serde(rename_all = "PascalCase")]
+#[serde(deny_unknown_fields)]
+pub struct EventChannel {
+ /// Transport kind of the channel.
+ pub kind: EventChannelKind,
+
+ /// Transport-specific path. For the `LocalPipe` kind this is the pipe
+ /// name/path the client should connect to.
+ #[schemars(length(min = 1, max = 1024))]
+ pub path: String,
+}
+
+/// Event channel protocol major version advertised in the `Hello` frame.
+pub const EVENT_CHANNEL_VERSION_MAJOR: u16 = 1;
+/// Event channel protocol minor version advertised in the `Hello` frame.
+pub const EVENT_CHANNEL_VERSION_MINOR: u16 = 0;
+
+/// Size in bytes of the fixed frame header (`u32 frame_size` + `u16 frame_kind`).
+pub const EVENT_FRAME_HEADER_SIZE: usize = 6;
+
+/// Maximum allowed frame body size in bytes.
+///
+/// Protects decoders from unbounded buffering; encoders must split larger
+/// output into multiple frames.
+pub const MAX_EVENT_FRAME_BODY_BYTES: usize = 65536;
+
+/// `NOW_BROKER` event frame kind discriminators.
+pub mod frame_kind {
+ /// `NOW_BROKER_HELLO`
+ pub const HELLO: u16 = 0x0000;
+ /// `NOW_BROKER_STATUS_UPDATED`
+ pub const STATUS_UPDATED: u16 = 0x0001;
+ /// `NOW_BROKER_FINISH`
+ pub const FINISH: u16 = 0x0002;
+ /// `NOW_BROKER_STDOUT`
+ pub const STDOUT: u16 = 0x0003;
+ /// `NOW_BROKER_STDERR`
+ pub const STDERR: u16 = 0x0004;
+ /// `NOW_BROKER_STDOUT_OVERFLOW`
+ pub const STDOUT_OVERFLOW: u16 = 0x0005;
+ /// `NOW_BROKER_STDERR_OVERFLOW`
+ pub const STDERR_OVERFLOW: u16 = 0x0006;
+}
+
+/// A decoded `NOW_BROKER` event frame.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum EventFrame {
+ /// First frame on the channel; acknowledges that the transport is ready
+ /// and advertises the protocol version.
+ Hello { version_major: u16, version_minor: u16 },
+ /// The operation status changed and awaits a `StatusRequest` HTTP query.
+ /// Deliberately carries no payload: complex data is queried over HTTP.
+ StatusUpdated,
+ /// New stdout data. Strictly UTF-8; character boundaries are guaranteed
+ /// by the broker.
+ Stdout(String),
+ /// New stderr data. Strictly UTF-8; character boundaries are guaranteed
+ /// by the broker.
+ Stderr(String),
+ /// The operation finished; query `StatusRequest` for details. The channel
+ /// can be gracefully closed by the client after this frame.
+ Finish,
+ /// The client was too slow to read stdout and `bytes_skipped` bytes were
+ /// truncated.
+ StdoutOverflow { bytes_skipped: u32 },
+ /// The client was too slow to read stderr and `bytes_skipped` bytes were
+ /// truncated.
+ StderrOverflow { bytes_skipped: u32 },
+ /// A frame with an unknown kind. Must be ignored by consumers to allow
+ /// forward-compatible protocol extension.
+ Unknown { kind: u16, body: Vec },
+}
+
+/// Error produced while encoding or decoding event frames.
+#[derive(Debug, thiserror::Error, PartialEq, Eq)]
+pub enum EventFrameError {
+ #[error("frame body size {size} exceeds the maximum of {MAX_EVENT_FRAME_BODY_BYTES} bytes")]
+ BodyTooLarge { size: usize },
+ #[error("frame kind {kind:#06x} expects a body of {expected} bytes, got {actual}")]
+ InvalidBodyLength { kind: u16, expected: usize, actual: usize },
+ #[error("stdout/stderr frame body is not valid UTF-8")]
+ InvalidUtf8,
+}
+
+impl EventFrame {
+ /// Frame kind discriminator for this frame.
+ pub fn kind(&self) -> u16 {
+ match self {
+ EventFrame::Hello { .. } => frame_kind::HELLO,
+ EventFrame::StatusUpdated => frame_kind::STATUS_UPDATED,
+ EventFrame::Stdout(_) => frame_kind::STDOUT,
+ EventFrame::Stderr(_) => frame_kind::STDERR,
+ EventFrame::Finish => frame_kind::FINISH,
+ EventFrame::StdoutOverflow { .. } => frame_kind::STDOUT_OVERFLOW,
+ EventFrame::StderrOverflow { .. } => frame_kind::STDERR_OVERFLOW,
+ EventFrame::Unknown { kind, .. } => *kind,
+ }
+ }
+
+ /// Encode the frame (header + body) into a byte vector.
+ pub fn encode(&self) -> Result, EventFrameError> {
+ let body: Vec = match self {
+ EventFrame::Hello {
+ version_major,
+ version_minor,
+ } => {
+ let mut body = Vec::with_capacity(4);
+ body.extend_from_slice(&version_major.to_le_bytes());
+ body.extend_from_slice(&version_minor.to_le_bytes());
+ body
+ }
+ EventFrame::StatusUpdated | EventFrame::Finish => Vec::new(),
+ EventFrame::Stdout(data) | EventFrame::Stderr(data) => data.as_bytes().to_vec(),
+ EventFrame::StdoutOverflow { bytes_skipped } | EventFrame::StderrOverflow { bytes_skipped } => {
+ bytes_skipped.to_le_bytes().to_vec()
+ }
+ EventFrame::Unknown { body, .. } => body.clone(),
+ };
+
+ if body.len() > MAX_EVENT_FRAME_BODY_BYTES {
+ return Err(EventFrameError::BodyTooLarge { size: body.len() });
+ }
+
+ let mut frame = Vec::with_capacity(EVENT_FRAME_HEADER_SIZE + body.len());
+ frame.extend_from_slice(&u32::try_from(body.len()).expect("bounded above").to_le_bytes());
+ frame.extend_from_slice(&self.kind().to_le_bytes());
+ frame.extend_from_slice(&body);
+ Ok(frame)
+ }
+
+ /// Decode a frame from a kind discriminator and body bytes.
+ pub fn decode_body(kind: u16, body: &[u8]) -> Result {
+ fn expect_len(kind: u16, body: &[u8], expected: usize) -> Result<(), EventFrameError> {
+ if body.len() != expected {
+ return Err(EventFrameError::InvalidBodyLength {
+ kind,
+ expected,
+ actual: body.len(),
+ });
+ }
+ Ok(())
+ }
+
+ match kind {
+ frame_kind::HELLO => {
+ expect_len(kind, body, 4)?;
+ Ok(EventFrame::Hello {
+ version_major: u16::from_le_bytes([body[0], body[1]]),
+ version_minor: u16::from_le_bytes([body[2], body[3]]),
+ })
+ }
+ frame_kind::STATUS_UPDATED => {
+ expect_len(kind, body, 0)?;
+ Ok(EventFrame::StatusUpdated)
+ }
+ frame_kind::STDOUT => {
+ let data = str::from_utf8(body).map_err(|_| EventFrameError::InvalidUtf8)?;
+ Ok(EventFrame::Stdout(data.to_owned()))
+ }
+ frame_kind::STDERR => {
+ let data = str::from_utf8(body).map_err(|_| EventFrameError::InvalidUtf8)?;
+ Ok(EventFrame::Stderr(data.to_owned()))
+ }
+ frame_kind::FINISH => {
+ expect_len(kind, body, 0)?;
+ Ok(EventFrame::Finish)
+ }
+ frame_kind::STDOUT_OVERFLOW => {
+ expect_len(kind, body, 4)?;
+ Ok(EventFrame::StdoutOverflow {
+ bytes_skipped: u32::from_le_bytes([body[0], body[1], body[2], body[3]]),
+ })
+ }
+ frame_kind::STDERR_OVERFLOW => {
+ expect_len(kind, body, 4)?;
+ Ok(EventFrame::StderrOverflow {
+ bytes_skipped: u32::from_le_bytes([body[0], body[1], body[2], body[3]]),
+ })
+ }
+ _ => Ok(EventFrame::Unknown {
+ kind,
+ body: body.to_vec(),
+ }),
+ }
+ }
+}
+
+/// Incremental decoder for a `NOW_BROKER` event frame stream.
+///
+/// Feed raw bytes read from the channel with [`EventFrameDecoder::extend`],
+/// then drain complete frames with [`EventFrameDecoder::next_frame`].
+#[derive(Debug, Default)]
+pub struct EventFrameDecoder {
+ buffer: Vec,
+}
+
+impl EventFrameDecoder {
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ /// Append raw bytes received from the transport.
+ pub fn extend(&mut self, bytes: &[u8]) {
+ self.buffer.extend_from_slice(bytes);
+ }
+
+ /// Returns `true` when the decoder holds buffered bytes that do not yet
+ /// form a complete frame. If the transport reaches end-of-stream while
+ /// this is `true`, the frame stream was truncated mid-frame.
+ pub fn has_buffered_data(&self) -> bool {
+ !self.buffer.is_empty()
+ }
+
+ /// Try to decode the next complete frame.
+ ///
+ /// Returns `Ok(None)` when more bytes are needed. Frames with unknown
+ /// kinds are returned as [`EventFrame::Unknown`] and should be ignored by
+ /// the consumer. Errors are not recoverable: the channel is corrupt and
+ /// should be closed.
+ pub fn next_frame(&mut self) -> Result, EventFrameError> {
+ if self.buffer.len() < EVENT_FRAME_HEADER_SIZE {
+ return Ok(None);
+ }
+
+ let size = u32::from_le_bytes([self.buffer[0], self.buffer[1], self.buffer[2], self.buffer[3]]) as usize;
+ if size > MAX_EVENT_FRAME_BODY_BYTES {
+ return Err(EventFrameError::BodyTooLarge { size });
+ }
+
+ let kind = u16::from_le_bytes([self.buffer[4], self.buffer[5]]);
+ let total = EVENT_FRAME_HEADER_SIZE + size;
+ if self.buffer.len() < total {
+ return Ok(None);
+ }
+
+ let frame = EventFrame::decode_body(kind, &self.buffer[EVENT_FRAME_HEADER_SIZE..total])?;
+ self.buffer.drain(..total);
+ Ok(Some(frame))
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ #![allow(clippy::unwrap_used)]
+
+ use super::*;
+
+ fn round_trip(frame: EventFrame) {
+ let bytes = frame.encode().unwrap();
+ let mut decoder = EventFrameDecoder::new();
+ decoder.extend(&bytes);
+ assert_eq!(decoder.next_frame().unwrap(), Some(frame));
+ assert_eq!(decoder.next_frame().unwrap(), None);
+ }
+
+ #[test]
+ fn frames_round_trip() {
+ round_trip(EventFrame::Hello {
+ version_major: EVENT_CHANNEL_VERSION_MAJOR,
+ version_minor: EVENT_CHANNEL_VERSION_MINOR,
+ });
+ round_trip(EventFrame::StatusUpdated);
+ round_trip(EventFrame::Stdout("hello π\n".to_owned()));
+ round_trip(EventFrame::Stderr("warning: π\n".to_owned()));
+ round_trip(EventFrame::Finish);
+ round_trip(EventFrame::StdoutOverflow { bytes_skipped: 4096 });
+ round_trip(EventFrame::StderrOverflow { bytes_skipped: 1 });
+ }
+
+ #[test]
+ fn hello_frame_has_documented_layout() {
+ let bytes = EventFrame::Hello {
+ version_major: 1,
+ version_minor: 0,
+ }
+ .encode()
+ .unwrap();
+ assert_eq!(bytes, [4, 0, 0, 0, 0x00, 0x00, 1, 0, 0, 0]);
+ }
+
+ #[test]
+ fn decoder_handles_partial_and_concatenated_input() {
+ let mut stream = Vec::new();
+ stream.extend_from_slice(
+ &EventFrame::Hello {
+ version_major: 1,
+ version_minor: 0,
+ }
+ .encode()
+ .unwrap(),
+ );
+ stream.extend_from_slice(&EventFrame::Stdout("chunk".to_owned()).encode().unwrap());
+ stream.extend_from_slice(&EventFrame::Finish.encode().unwrap());
+
+ let mut decoder = EventFrameDecoder::new();
+ let mut frames = Vec::new();
+ for byte in stream {
+ decoder.extend(&[byte]);
+ while let Some(frame) = decoder.next_frame().unwrap() {
+ frames.push(frame);
+ }
+ }
+
+ assert_eq!(
+ frames,
+ [
+ EventFrame::Hello {
+ version_major: 1,
+ version_minor: 0
+ },
+ EventFrame::Stdout("chunk".to_owned()),
+ EventFrame::Finish,
+ ]
+ );
+ }
+
+ #[test]
+ fn decoder_skips_unknown_frame_kinds() {
+ let unknown = EventFrame::Unknown {
+ kind: 0x7fff,
+ body: vec![1, 2, 3],
+ };
+ let mut decoder = EventFrameDecoder::new();
+ decoder.extend(&unknown.encode().unwrap());
+ decoder.extend(&EventFrame::Finish.encode().unwrap());
+
+ assert_eq!(decoder.next_frame().unwrap(), Some(unknown));
+ assert_eq!(decoder.next_frame().unwrap(), Some(EventFrame::Finish));
+ }
+
+ #[test]
+ fn decoder_rejects_invalid_utf8_and_oversized_frames() {
+ let mut decoder = EventFrameDecoder::new();
+ decoder.extend(&[2, 0, 0, 0, 0x03, 0x00, 0xff, 0xfe]);
+ assert_eq!(decoder.next_frame(), Err(EventFrameError::InvalidUtf8));
+
+ let mut decoder = EventFrameDecoder::new();
+ let oversized = (u32::try_from(MAX_EVENT_FRAME_BODY_BYTES).unwrap() + 1).to_le_bytes();
+ decoder.extend(&[oversized[0], oversized[1], oversized[2], oversized[3], 0x03, 0x00]);
+ assert!(matches!(
+ decoder.next_frame(),
+ Err(EventFrameError::BodyTooLarge { .. })
+ ));
+ }
+
+ #[test]
+ fn decoder_reports_buffered_data_for_truncated_frames() {
+ let mut decoder = EventFrameDecoder::new();
+ assert!(!decoder.has_buffered_data());
+
+ let bytes = EventFrame::Stdout("interrupted".to_owned()).encode().unwrap();
+ decoder.extend(&bytes[..bytes.len() - 4]);
+ assert_eq!(decoder.next_frame().unwrap(), None);
+ // EOF here would mean the stream was truncated mid-frame.
+ assert!(decoder.has_buffered_data());
+
+ decoder.extend(&bytes[bytes.len() - 4..]);
+ assert_eq!(
+ decoder.next_frame().unwrap(),
+ Some(EventFrame::Stdout("interrupted".to_owned()))
+ );
+ assert!(!decoder.has_buffered_data());
+ }
+
+ #[test]
+ fn fixed_body_frames_reject_wrong_length() {
+ assert_eq!(
+ EventFrame::decode_body(frame_kind::HELLO, &[1, 0]),
+ Err(EventFrameError::InvalidBodyLength {
+ kind: frame_kind::HELLO,
+ expected: 4,
+ actual: 2
+ })
+ );
+ assert_eq!(
+ EventFrame::decode_body(frame_kind::FINISH, &[0]),
+ Err(EventFrameError::InvalidBodyLength {
+ kind: frame_kind::FINISH,
+ expected: 0,
+ actual: 1
+ })
+ );
+ }
+}
diff --git a/policies/rust/now-policy-api/src/execute.rs b/policies/rust/now-policy-api/src/execute.rs
index d3e8dbf..a5c4e83 100644
--- a/policies/rust/now-policy-api/src/execute.rs
+++ b/policies/rust/now-policy-api/src/execute.rs
@@ -8,6 +8,7 @@ use serde::{Deserialize, Serialize};
use super::api::{DecisionInfo, OperationDiagnostics, RequestSummary, ResponsePolicyInfo, ServerContext};
use super::enums::OperationStatus;
+use super::event_channel::EventChannel;
use super::{ApiVersion, ExecutionResponseKind, ResourceId};
/// Response returned after an execute request is evaluated and, when allowed, submitted.
@@ -66,4 +67,11 @@ pub struct OperationSubmission {
/// UTC timestamp when the operation was accepted.
pub submitted_at: DateTime,
+
+ /// Per-operation event channel carrying `NOW_BROKER` event frames (status
+ /// change notifications and, when the execute request opted in via
+ /// `CaptureOutput`, stdout/stderr data). Present whenever the broker
+ /// supports event channels; absent otherwise.
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub event_channel: Option,
}
diff --git a/policies/rust/now-policy-api/src/lib.rs b/policies/rust/now-policy-api/src/lib.rs
index 9c3988a..f194863 100644
--- a/policies/rust/now-policy-api/src/lib.rs
+++ b/policies/rust/now-policy-api/src/lib.rs
@@ -8,6 +8,7 @@ pub mod cancel;
pub mod capabilities;
pub mod enums;
pub mod evaluate;
+pub mod event_channel;
pub mod execute;
pub mod health;
#[cfg(feature = "policy-compat")]
@@ -19,6 +20,7 @@ pub use cancel::*;
pub use capabilities::*;
pub use enums::*;
pub use evaluate::*;
+pub use event_channel::*;
pub use execute::*;
pub use health::*;
pub use status::*;
@@ -540,51 +542,3 @@ impl<'de> Deserialize<'de> for CommandString {
Self::parse(&s).map_err(serde::de::Error::custom)
}
}
-
-/// Base64-encoded UTF-8 operation output.
-#[derive(
- Debug, Clone, PartialEq, Eq, Serialize, JsonSchema, derive_more::AsRef, derive_more::Deref, derive_more::From,
-)]
-#[as_ref(str)]
-#[deref(forward)]
-pub struct Base64Utf8Data(#[schemars(length(max = 16384), regex(pattern = r"^[A-Za-z0-9+/]*={0,2}$"))] pub String);
-
-impl Base64Utf8Data {
- pub fn parse(s: &str) -> Result {
- if s.len() > 16384 {
- return Err(ModelValidationError::Invalid {
- type_name: "Base64Utf8Data",
- reason: format!("length {} exceeds maximum 16384", s.len()),
- });
- }
-
- use base64::Engine;
- let decoded =
- base64::engine::general_purpose::STANDARD
- .decode(s)
- .map_err(|e| ModelValidationError::Invalid {
- type_name: "Base64Utf8Data",
- reason: e.to_string(),
- })?;
-
- core::str::from_utf8(&decoded).map_err(|e| ModelValidationError::Invalid {
- type_name: "Base64Utf8Data",
- reason: e.to_string(),
- })?;
-
- Ok(Self(s.to_owned()))
- }
-}
-
-impl<'de> Deserialize<'de> for Base64Utf8Data {
- fn deserialize>(deserializer: D) -> Result {
- let s = String::deserialize(deserializer)?;
- Self::parse(&s).map_err(serde::de::Error::custom)
- }
-}
-
-impl From<&str> for Base64Utf8Data {
- fn from(s: &str) -> Self {
- Self(s.to_owned())
- }
-}
diff --git a/policies/rust/now-policy-api/src/status.rs b/policies/rust/now-policy-api/src/status.rs
index ee7a5ca..f229e1d 100644
--- a/policies/rust/now-policy-api/src/status.rs
+++ b/policies/rust/now-policy-api/src/status.rs
@@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize};
use super::api::{ClientContext, ServerContext};
use super::enums::OperationStatus;
-use super::{ApiVersion, Base64Utf8Data, ResourceId, StatusRequestKind, StatusResponseKind};
+use super::{ApiVersion, ResourceId, StatusRequestKind, StatusResponseKind};
/// Request body for querying an operation status.
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
@@ -72,9 +72,4 @@ pub struct StatusResponse {
/// Manager-specific structured status details.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub details: Option,
-
- /// Captured combined stdout+stderr as base64-encoded UTF-8 data (tail-truncated to ~10 KiB before encoding).
- /// Only present when the original request opted in via `CaptureOutput`.
- #[serde(default, skip_serializing_if = "Option::is_none")]
- pub stdout: Option,
}
diff --git a/policies/rust/now-policy-server-template/assets/samples/frames/event-channel.frames.bin b/policies/rust/now-policy-server-template/assets/samples/frames/event-channel.frames.bin
new file mode 100644
index 0000000..e4f54b2
Binary files /dev/null and b/policies/rust/now-policy-server-template/assets/samples/frames/event-channel.frames.bin differ
diff --git a/policies/rust/now-policy-server-template/assets/samples/responses/execution-winget-vscode-install.response.json b/policies/rust/now-policy-server-template/assets/samples/responses/execution-winget-vscode-install.response.json
index a8ebbaf..feea983 100644
--- a/policies/rust/now-policy-server-template/assets/samples/responses/execution-winget-vscode-install.response.json
+++ b/policies/rust/now-policy-server-template/assets/samples/responses/execution-winget-vscode-install.response.json
@@ -23,7 +23,11 @@
"Operation": {
"OperationId": "op-winget-vscode-install-000001",
"Status": "Starting",
- "SubmittedAt": "2026-05-05T12:00:01Z"
+ "SubmittedAt": "2026-05-05T12:00:01Z",
+ "EventChannel": {
+ "Kind": "LocalPipe",
+ "Path": "Devolutions.Now.PackageBroker.Operation.op-winget-vscode-install-000001"
+ }
},
"Diagnostics": {
"CommandPreview": [
diff --git a/policies/rust/now-policy-server-template/assets/samples/responses/status-failed.response.json b/policies/rust/now-policy-server-template/assets/samples/responses/status-failed.response.json
index faec4a8..40ed9cc 100644
--- a/policies/rust/now-policy-server-template/assets/samples/responses/status-failed.response.json
+++ b/policies/rust/now-policy-server-template/assets/samples/responses/status-failed.response.json
@@ -7,7 +7,6 @@
"StartedAt": "2026-05-05T12:01:00Z",
"CompletedAt": "2026-05-05T12:01:05Z",
"ExitCode": 1,
- "Stdout": "Rm91bmQgR2l0IFtHaXQuR2l0XQpUaGlzIHBhY2thZ2UncyBpbnN0YWxsZXIgcmVsaWVzIG9uIHRoZSBmb2xsb3dpbmcgZGVwZW5kZW5jaWVzLi4uClVuaW5zdGFsbCBmYWlsZWQgd2l0aCBleGl0IGNvZGU6IDE=",
"Server": {
"Transport": "HttpNamedPipe",
"ServerVersion": "0.1.0"
diff --git a/policies/rust/now-policy-server-template/tests/sample_documents.rs b/policies/rust/now-policy-server-template/tests/sample_documents.rs
index 9ee8ea5..8d6a0e6 100644
--- a/policies/rust/now-policy-server-template/tests/sample_documents.rs
+++ b/policies/rust/now-policy-server-template/tests/sample_documents.rs
@@ -179,6 +179,44 @@ fn request_kind_marker_rejects_wrong_value() {
);
}
+#[test]
+fn event_channel_frame_sample_decodes() {
+ use now_policy_server_template::{EventFrame, EventFrameDecoder};
+
+ let bytes = std::fs::read(samples_dir().join("frames/event-channel.frames.bin")).unwrap();
+
+ let mut decoder = EventFrameDecoder::new();
+ decoder.extend(&bytes);
+ let mut frames = Vec::new();
+ while let Some(frame) = decoder.next_frame().unwrap() {
+ frames.push(frame);
+ }
+
+ assert_eq!(
+ frames,
+ [
+ EventFrame::Hello {
+ version_major: 1,
+ version_minor: 0
+ },
+ EventFrame::StatusUpdated,
+ EventFrame::Stdout("hello \u{03c0}\n".to_owned()),
+ EventFrame::Stderr("oops\n".to_owned()),
+ EventFrame::StdoutOverflow { bytes_skipped: 4096 },
+ EventFrame::StderrOverflow { bytes_skipped: 16 },
+ EventFrame::Unknown {
+ kind: 0x7fff,
+ body: vec![1, 2, 3]
+ },
+ EventFrame::Finish,
+ ]
+ );
+
+ // The fixture must round-trip byte-for-byte through the encoder.
+ let reencoded: Vec = frames.iter().flat_map(|f| f.encode().unwrap()).collect();
+ assert_eq!(reencoded, bytes);
+}
+
#[tokio::test]
async fn mock_server_returns_registered_fixture_responses() {
let request_path = samples_dir().join("requests/winget-vscode-install.request.json");
@@ -205,6 +243,13 @@ async fn mock_server_returns_registered_fixture_responses() {
execution.operation.as_ref().map(|op| &op.operation_id)
);
+ let event_channel = executed.operation.unwrap().event_channel.unwrap();
+ assert_eq!(
+ event_channel.kind,
+ now_policy_server_template::EventChannelKind::LocalPipe
+ );
+ assert!(!event_channel.path.is_empty());
+
let status_request = StatusRequest {
request_kind: StatusRequestKind,
request_version: API_VERSION_STR.into(),