Agent output from a remote executor is not captured, so live supervision and post-hoc stream diagnosis are lost for remote phases

docs/operating/remote-executors.md

@@ -43,7 +43,23 @@ after failure stays with the pipeline state machine.
 Bounds live under `CodeyBox:ExecutorPhaseDispatch` (`StageOutMaxArchiveBytes`,
 `StageOutMaxEntries`, `StageOutMaxExpansionRatio`, `IdempotencyTtl`,
 `MaxRequestPayloadBytes`, `MaxResultFindings`, `MaxFindingLengthChars`,
-`MaxResultErrorLengthChars`), hot-reloadable like the other dispatch knobs.
+`MaxResultErrorLengthChars`, `MaxStreamChunkChars`), hot-reloadable like the other dispatch knobs.
+
+## Live agent-output relay
+
+While the phase runs, the executor streams sequenced agent-output chunks
+(`ExecutorStreamChunk`, numbered from zero) back to the orchestrator as they
+are produced — transports implementing `IStreamingExecutorPhaseTransport`
+deliver them live rather than buffering to phase end. The proxy relays each
+chunk into the orchestrator-side stream capture at the same path and key
+(work-item directory, phase/iteration file) a local phase would write, and
+re-broadcasts it through the existing stdout hub, so live subscribers see
+remote output with no contract change. The relay holds no queue of its own:
+the capture's own slicing and per-file truncation (including its truncation
+marker) apply unchanged, and `MaxStreamChunkChars` only caps the size of a
+single forwarded piece. A lost or reordered chunk is recorded as an explicit
+`[...stream gap ...]` line rather than silently omitted, and relay failure
+never fails the phase — losing the stream degrades observability only.
 
 ## Running the executor
 

src/CodeyBox.Core/ExecutorPhaseTransport.cs

@@ -119,6 +119,34 @@ public interface IExecutorPhaseTransport
     Task StageOutToArchiveAsync(string hostArchivePath, long maxArchiveBytes, CancellationToken ct);
 }
 
+/// <summary>
+/// Streaming extension to <see cref="IExecutorPhaseTransport"/>: the executor
+/// delivers live agent-output chunks to <paramref name="onChunk"/> as they
+/// are produced — implementations MUST NOT buffer to phase end, since the
+/// orchestrator feeds them into the live capture and the supervision hub in
+/// real time (the same requirement <c>IRemoteHostTransport</c> documents for
+/// remote agent CLIs).
+///
+/// <para>Chunks are numbered from zero with no gaps (see
+/// <see cref="ExecutorStreamChunk"/>). The callback itself never fails the
+/// phase: the orchestrator-side relay swallows its own failures, and the
+/// transport must not treat a callback exception as a phase or transport
+/// failure — losing the stream degrades observability only.</para>
+/// </summary>
+public interface IStreamingExecutorPhaseTransport : IExecutorPhaseTransport
+{
+    /// <summary>
+    /// Runs the phase like <see cref="IExecutorPhaseTransport.RunPhaseAsync"/>
+    /// while invoking <paramref name="onChunk"/> for each output chunk in
+    /// emission order as it is produced. A null callback behaves exactly like
+    /// the non-streaming overload.
+    /// </summary>
+    Task<ExecutorPhaseResult> RunPhaseAsync(
+        ExecutorPhaseRequest request,
+        Func<ExecutorStreamChunk, CancellationToken, Task>? onChunk,
+        CancellationToken ct);
+}
+
 /// <summary>
 /// Resolves the dispatch transport for a registered executor host. Returns
 /// null when the host has no transport configured (for example no SSH target

src/CodeyBox.Core/ExecutorStreamChunk.cs

@@ -0,0 +1,34 @@
+using System.Text.Json.Serialization;
+
+namespace CodeyBox.Core;
+
+/// <summary>
+/// One live agent-output chunk produced on an executor host and relayed to
+/// the orchestrator while the phase runs. The orchestrator appends
+/// <see cref="Data"/> to the same <c>AgentStreamCapture</c> artefact (same
+/// directory, same phase/iteration key) a local phase would write, and
+/// re-broadcasts it through the existing stdout hub, so live subscribers see
+/// remote output with no contract change.
+///
+/// <para>Sequencing: the executor numbers chunks from zero with no gaps.
+/// When the relay observes a discontinuity (a lost or reordered chunk) it
+/// records an explicit gap marker in the captured stream rather than
+/// presenting a contiguous stream that silently omits output.</para>
+/// </summary>
+public sealed record ExecutorStreamChunk
+{
+    /// <summary>
+    /// Zero-based position of this chunk in the executor's emission order.
+    /// Must be zero or positive; the first chunk of a dispatch is zero.
+    /// </summary>
+    [JsonPropertyName("sequence")]
+    public required long Sequence { get; init; }
+
+    /// <summary>
+    /// Raw agent-output text for this chunk. May be an arbitrary slice of
+    /// the stream (line fragments are fine — the capture reassembles lines).
+    /// Null is treated as empty by the relay.
+    /// </summary>
+    [JsonPropertyName("data")]
+    public string? Data { get; init; }
+}

src/CodeyBox.Orchestrator/ExecutorPhaseDispatchOptions.cs

@@ -50,6 +50,16 @@ public sealed class ExecutorPhaseDispatchOptions
     /// <summary>Maximum chars accepted in an executor-returned error message.</summary>
     public int MaxResultErrorLengthChars { get; set; } = 8192;
 
+    /// <summary>
+    /// Largest single piece of relayed stream text forwarded to the capture
+    /// and the live broadcast per write. Larger executor payloads are split
+    /// into pieces of at most this size before forwarding, mirroring the
+    /// capture's own queue slicing, so one hostile chunk cannot force an
+    /// unbounded single allocation through the relay. The per-file size cap
+    /// that truncates with a marker stays owned by the capture itself.
+    /// </summary>
+    public int MaxStreamChunkChars { get; set; } = 64 * 1024;
+
     /// <summary>
     /// Fails fast on misconfiguration so a bad bound surfaces at dispatch
     /// time instead of silently admitting an unbounded payload.
@@ -80,5 +90,8 @@ public void Validate()
         if (MaxResultErrorLengthChars <= 0)
             throw new InvalidOperationException(
                 "CodeyBox:ExecutorPhaseDispatch:MaxResultErrorLengthChars must be > 0.");
+        if (MaxStreamChunkChars <= 0)
+            throw new InvalidOperationException(
+                "CodeyBox:ExecutorPhaseDispatch:MaxStreamChunkChars must be > 0.");
     }
 }

src/CodeyBox.Orchestrator/ExecutorPhaseProxy.cs

@@ -27,6 +27,12 @@ namespace CodeyBox.Orchestrator;
 /// <item>Stage the phase's single bare repo to the executor, run the phase
 /// there, and stage the repo back as a tar archive. Only the per-item repo
 /// path is ever transferred — never the whole repos root.</item>
+/// <item>While the phase runs, relay its live agent-output chunks into the
+/// orchestrator-side stream capture (same directory, same phase/iteration
+/// key a local phase would write) and the existing stdout broadcast, so a
+/// remote phase leaves the same observable artefact as a local one. Relay
+/// failure never fails the phase; a lost or reordered chunk is recorded as
+/// an explicit gap marker, never silently omitted.</item>
 /// <item>Validate the staged-back archive (size, entry count, expansion
 /// ratio, path containment) and install it over the bare repo. Violations
 /// fail the phase without writing to the bare repo and without caching.</item>
@@ -56,6 +62,8 @@ public sealed class ExecutorPhaseProxy : IExecutorPhaseRunner
     private readonly IIdempotencyStore _idempotency;
     private readonly IExecutorPhaseRunner _inner;
     private readonly Func<ExecutorPhaseDispatchOptions> _optionsAccessor;
+    private readonly IAgentStreamStore? _streamStore;
+    private readonly IStdoutBroadcaster? _broadcaster;
     private readonly TimeProvider _clock;
     private readonly ILogger<ExecutorPhaseProxy> _log;
 
@@ -67,7 +75,9 @@ public ExecutorPhaseProxy(
         IExecutorPhaseRunner inner,
         Func<ExecutorPhaseDispatchOptions> optionsAccessor,
         TimeProvider? clock = null,
-        ILogger<ExecutorPhaseProxy>? log = null)
+        ILogger<ExecutorPhaseProxy>? log = null,
+        IAgentStreamStore? streamStore = null,
+        IStdoutBroadcaster? broadcaster = null)
     {
         _registry = registry ?? throw new ArgumentNullException(nameof(registry));
         _transports = transports ?? throw new ArgumentNullException(nameof(transports));
@@ -77,6 +87,8 @@ public ExecutorPhaseProxy(
         _optionsAccessor = optionsAccessor ?? throw new ArgumentNullException(nameof(optionsAccessor));
         _clock = clock ?? TimeProvider.System;
         _log = log ?? NullLogger<ExecutorPhaseProxy>.Instance;
+        _streamStore = streamStore;
+        _broadcaster = broadcaster;
     }
 
     public async Task<ExecutorPhaseResult> ExecutePhaseAsync(ExecutorPhaseRequest request, CancellationToken ct)
@@ -146,7 +158,57 @@ private async Task<ExecutorPhaseResult> ExecuteRemoteAsync(
 
         await CallTransportAsync(hostId, "stage-in", token => transport.StageInAsync(canonicalRepo, token), ct).ConfigureAwait(false);
 
-        var raw = await CallTransportAsync(hostId, "run-phase", token => transport.RunPhaseAsync(request, token), ct).ConfigureAwait(false);
+        // Open the orchestrator-side capture before the phase runs so a
+        // remote dispatch lands at the same path and key (work-item
+        // directory, phase/iteration file prefix) as the equivalent local
+        // phase. Attempt is zero-based while stream iterations are
+        // one-based, hence the +1. A request whose id is not a work-item
+        // GUID simply runs without a relayed stream.
+        AgentStreamCapture? streamCapture = null;
+        ExecutorStreamRelay? relay = null;
+        if (TryResolveStreamKey(request, out var streamWorkItem, out var streamIteration))
+        {
+            streamCapture = await BeginStreamCaptureAsync(streamWorkItem, request.Phase, streamIteration, ct).ConfigureAwait(false);
+            if (streamCapture is not null || _broadcaster is not null)
+                relay = new ExecutorStreamRelay(streamCapture, _broadcaster, streamWorkItem, request.Phase, _optionsAccessor, _log);
+        }
+
+        ExecutorPhaseResult raw;
+        try
+        {
+            if (relay is not null && transport is IStreamingExecutorPhaseTransport streaming)
+            {
+                var callback = relay.OnChunkAsync;
+                raw = await CallTransportAsync(
+                    hostId,
+                    "run-phase",
+                    token => streaming.RunPhaseAsync(request, callback, token),
+                    ct).ConfigureAwait(false);
+            }
+            else
+            {
+                raw = await CallTransportAsync(hostId, "run-phase", token => transport.RunPhaseAsync(request, token), ct).ConfigureAwait(false);
+            }
+        }
+        finally
+        {
+            // The artefact is complete once the phase returns: disposing
+            // flushes buffered chunks and records truncation exactly like a
+            // local phase, even when stage-out or validation fails next.
+            // Disposal never throws out of a remote dispatch.
+            if (streamCapture is not null)
+            {
+                try
+                {
+                    await streamCapture.DisposeAsync().ConfigureAwait(false);
+                }
+                catch (Exception ex)
+                {
+                    _log.LogDebug(ex, "Executor stream capture disposal failed on host {HostId}", hostId);
+                }
+            }
+        }
+
         var result = ValidateResult(raw, options);
 
         var scratchRoot = Path.Combine(Path.GetTempPath(), "codeybox-executor-phase-" + Guid.NewGuid().ToString("N"));
@@ -188,6 +250,38 @@ private async Task<IExecutorPhaseTransport> ResolveTransportAsync(string hostId,
         return transport;
     }
 
+    private static bool TryResolveStreamKey(ExecutorPhaseRequest request, out WorkItemId workItemId, out int iteration)
+    {
+        workItemId = default;
+        iteration = 0;
+        if (!Guid.TryParse(request.WorkItemId, out var guid))
+            return false;
+        workItemId = new WorkItemId(guid);
+        iteration = request.Attempt == int.MaxValue ? int.MaxValue : request.Attempt + 1;
+        return iteration >= 1;
+    }
+
+    private async Task<AgentStreamCapture?> BeginStreamCaptureAsync(
+        WorkItemId workItemId,
+        string phase,
+        int iteration,
+        CancellationToken ct)
+    {
+        if (_streamStore is null)
+            return null;
+        try
+        {
+            return await _streamStore.BeginCaptureAsync(workItemId, phase, iteration, ct).ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            // Losing the stream degrades observability; it is never an agent
+            // or work-item failure.
+            _log.LogDebug(ex, "Executor stream capture unavailable for phase {Phase}", phase);
+            return null;
+        }
+    }
+
     private static async Task<T> CallTransportAsync<T>(
         string hostId,
         string operation,