diff --git a/docs/operating/remote-executors.md b/docs/operating/remote-executors.md index 5071f47b..9c5fc3e3 100644 --- a/docs/operating/remote-executors.md +++ b/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 diff --git a/src/CodeyBox.Core/ExecutorPhaseTransport.cs b/src/CodeyBox.Core/ExecutorPhaseTransport.cs index 1303c39f..8df8834f 100644 --- a/src/CodeyBox.Core/ExecutorPhaseTransport.cs +++ b/src/CodeyBox.Core/ExecutorPhaseTransport.cs @@ -119,6 +119,34 @@ public interface IExecutorPhaseTransport Task StageOutToArchiveAsync(string hostArchivePath, long maxArchiveBytes, CancellationToken ct); } +/// +/// Streaming extension to : the executor +/// delivers live agent-output chunks to 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 IRemoteHostTransport documents for +/// remote agent CLIs). +/// +/// Chunks are numbered from zero with no gaps (see +/// ). 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. +/// +public interface IStreamingExecutorPhaseTransport : IExecutorPhaseTransport +{ + /// + /// Runs the phase like + /// while invoking for each output chunk in + /// emission order as it is produced. A null callback behaves exactly like + /// the non-streaming overload. + /// + Task RunPhaseAsync( + ExecutorPhaseRequest request, + Func? onChunk, + CancellationToken ct); +} + /// /// Resolves the dispatch transport for a registered executor host. Returns /// null when the host has no transport configured (for example no SSH target diff --git a/src/CodeyBox.Core/ExecutorStreamChunk.cs b/src/CodeyBox.Core/ExecutorStreamChunk.cs new file mode 100644 index 00000000..59b070e7 --- /dev/null +++ b/src/CodeyBox.Core/ExecutorStreamChunk.cs @@ -0,0 +1,34 @@ +using System.Text.Json.Serialization; + +namespace CodeyBox.Core; + +/// +/// One live agent-output chunk produced on an executor host and relayed to +/// the orchestrator while the phase runs. The orchestrator appends +/// to the same AgentStreamCapture 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. +/// +/// 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. +/// +public sealed record ExecutorStreamChunk +{ + /// + /// 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. + /// + [JsonPropertyName("sequence")] + public required long Sequence { get; init; } + + /// + /// 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. + /// + [JsonPropertyName("data")] + public string? Data { get; init; } +} diff --git a/src/CodeyBox.Orchestrator/ExecutorPhaseDispatchOptions.cs b/src/CodeyBox.Orchestrator/ExecutorPhaseDispatchOptions.cs index 8d9051fc..f478aad7 100644 --- a/src/CodeyBox.Orchestrator/ExecutorPhaseDispatchOptions.cs +++ b/src/CodeyBox.Orchestrator/ExecutorPhaseDispatchOptions.cs @@ -50,6 +50,16 @@ public sealed class ExecutorPhaseDispatchOptions /// Maximum chars accepted in an executor-returned error message. public int MaxResultErrorLengthChars { get; set; } = 8192; + /// + /// 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. + /// + public int MaxStreamChunkChars { get; set; } = 64 * 1024; + /// /// 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."); } } diff --git a/src/CodeyBox.Orchestrator/ExecutorPhaseProxy.cs b/src/CodeyBox.Orchestrator/ExecutorPhaseProxy.cs index 0cb5d11b..f684ed6a 100644 --- a/src/CodeyBox.Orchestrator/ExecutorPhaseProxy.cs +++ b/src/CodeyBox.Orchestrator/ExecutorPhaseProxy.cs @@ -27,6 +27,12 @@ namespace CodeyBox.Orchestrator; /// 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. +/// 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. /// 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. @@ -56,6 +62,8 @@ public sealed class ExecutorPhaseProxy : IExecutorPhaseRunner private readonly IIdempotencyStore _idempotency; private readonly IExecutorPhaseRunner _inner; private readonly Func _optionsAccessor; + private readonly IAgentStreamStore? _streamStore; + private readonly IStdoutBroadcaster? _broadcaster; private readonly TimeProvider _clock; private readonly ILogger _log; @@ -67,7 +75,9 @@ public ExecutorPhaseProxy( IExecutorPhaseRunner inner, Func optionsAccessor, TimeProvider? clock = null, - ILogger? log = null) + ILogger? 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.Instance; + _streamStore = streamStore; + _broadcaster = broadcaster; } public async Task ExecutePhaseAsync(ExecutorPhaseRequest request, CancellationToken ct) @@ -146,7 +158,57 @@ private async Task 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 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 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 CallTransportAsync( string hostId, string operation, diff --git a/src/CodeyBox.Orchestrator/ExecutorStreamRelay.cs b/src/CodeyBox.Orchestrator/ExecutorStreamRelay.cs new file mode 100644 index 00000000..1adb8756 --- /dev/null +++ b/src/CodeyBox.Orchestrator/ExecutorStreamRelay.cs @@ -0,0 +1,145 @@ +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using CodeyBox.Core; + +namespace CodeyBox.Orchestrator; + +/// +/// Forwards live agent-output chunks produced on a remote executor host into +/// the orchestrator-side observability path: the same +/// artefact (same directory, same +/// phase/iteration key) a local phase would write, plus the existing +/// so live hub subscribers see remote +/// output with no contract change. +/// +/// The relay holds no queue of its own — every piece is forwarded +/// synchronously to the capture and the broadcaster — so a remote producer +/// cannot exceed the local buffering limits: the capture's own queue slicing +/// and per-file truncation (including its truncation marker) apply unchanged. +/// The only bound owned here is MaxStreamChunkChars, which caps the +/// size of a single forwarded piece before it reaches either sink. +/// +/// Sequencing: the executor numbers chunks from zero with no gaps. Any +/// discontinuity (lost or reordered chunk) is recorded as an explicit +/// [...stream gap ...] line in both the file and the broadcast rather +/// than presenting a contiguous stream that silently omits output. +/// +/// Failure isolation: relaying never fails the phase. This method never +/// throws — a broken capture or broadcaster only degrades observability. +/// +public sealed class ExecutorStreamRelay +{ + private readonly AgentStreamCapture? _capture; + private readonly IStdoutBroadcaster? _broadcaster; + private readonly WorkItemId _workItemId; + private readonly string _phase; + private readonly Func _optionsAccessor; + private readonly ILogger _log; + private readonly object _lock = new(); + private long _expectedSequence; + + public ExecutorStreamRelay( + AgentStreamCapture? capture, + IStdoutBroadcaster? broadcaster, + WorkItemId workItemId, + string phase, + Func optionsAccessor, + ILogger? log = null) + { + _capture = capture; + _broadcaster = broadcaster; + _workItemId = workItemId; + _phase = phase; + _optionsAccessor = optionsAccessor ?? throw new ArgumentNullException(nameof(optionsAccessor)); + _log = log ?? NullLogger.Instance; + } + + /// + /// Forwards one executor chunk to the capture and the live broadcast. + /// Records a gap marker when + /// is not the next expected value. Never throws. + /// + public Task OnChunkAsync(ExecutorStreamChunk chunk, CancellationToken ct) + { + try + { + if (chunk is null) + return Task.CompletedTask; + if (chunk.Sequence < 0) + { + _log.LogDebug( + "Ignoring executor stream chunk with negative sequence for phase {Phase}", + _phase); + return Task.CompletedTask; + } + + var data = chunk.Data ?? string.Empty; + var split = ReadSplitSize(); + lock (_lock) + { + if (chunk.Sequence != _expectedSequence) + { + Forward( + $"[...stream gap: expected seq {_expectedSequence} but received seq {chunk.Sequence}]\n", + split); + _expectedSequence = Math.Max(_expectedSequence, chunk.Sequence + 1); + } + else + { + _expectedSequence++; + } + + if (data.Length > 0) + Forward(data, split); + } + } + catch (Exception ex) + { + // Observability only: sequence numbers are safe to log, chunk + // payloads are untrusted executor output and never enter logs. + _log.LogDebug(ex, "Executor stream relay failed for phase {Phase}", _phase); + } + + return Task.CompletedTask; + } + + private int ReadSplitSize() + { + try + { + return Math.Max(1, _optionsAccessor().MaxStreamChunkChars); + } + catch (Exception ex) + { + // Hot-reload accessor failure must not break the relay; the 64 KiB + // floor mirrors the capture's own queue slice. + _log.LogDebug(ex, "Executor stream options unavailable for phase {Phase}", _phase); + return 64 * 1024; + } + } + + private void Forward(string text, int split) + { + for (var offset = 0; offset < text.Length; offset += split) + { + var piece = text.Substring(offset, Math.Min(split, text.Length - offset)); + try + { + _capture?.WriteChunk(piece); + } + catch (Exception ex) + { + _log.LogDebug(ex, "Executor stream capture write failed for phase {Phase}", _phase); + } + + try + { + _broadcaster?.BroadcastChunk(_workItemId, _phase, piece); + } + catch (Exception ex) + { + _log.LogDebug(ex, "Executor stream broadcast failed for phase {Phase}", _phase); + } + } + } +} diff --git a/tests/CodeyBox.Tests/ExecutorStreamRelayTests.cs b/tests/CodeyBox.Tests/ExecutorStreamRelayTests.cs new file mode 100644 index 00000000..676a20e8 --- /dev/null +++ b/tests/CodeyBox.Tests/ExecutorStreamRelayTests.cs @@ -0,0 +1,492 @@ +using System.Formats.Tar; +using System.Text; +using Microsoft.Extensions.Logging.Abstractions; +using CodeyBox.Core; +using CodeyBox.Git; +using CodeyBox.Orchestrator; + +namespace CodeyBox.Tests; + +/// +/// Verification for the executor stream relay: a phase run on a remote +/// executor produces a captured stream on the orchestrator at the same path +/// and key as the equivalent local phase, live subscribers receive chunks +/// during execution, over-limit producers are truncated with the same marker +/// a local producer receives, relay failure never changes the phase outcome, +/// and sequence gaps are recorded instead of silently omitted. Uses a real +/// , a real , +/// a real , a real +/// and real tar archives; only the +/// network hop to the executor is faked. +/// +public sealed class ExecutorStreamRelayTests : IDisposable +{ + private readonly string _root = Directory.CreateTempSubdirectory("codeybox-exec-relay-").FullName; + + public void Dispose() + { + try { Directory.Delete(_root, recursive: true); } catch { } + } + + // ── verification 1: same artefact, same path and key ──────────────────── + + [Fact] + public async Task RemotePhase_ProducesCapturedStream_AtSamePathAndKeyAsLocalPhase() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + using var ctx = CreateContext(); + var item = WorkItemId.New(); + await ctx.Git.EnsureRepositoryAsync(item, seedFromUrl: null, cts.Token); + ctx.Transport.Script = _ => [(0, "alpha\n"), (1, "beta\n")]; + + var result = await ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0), cts.Token); + + Assert.Equal(ExecutorPhaseOutcome.Succeeded, result.Outcome); + var file = Assert.Single(await ctx.Streams.ListAsync(item, ct: cts.Token)); + Assert.Equal("work", file.Phase); + Assert.Equal(1, file.Iteration); + Assert.Matches(@"^work-1-[0-9a-f]{6}\.jsonl$", file.FileName); + Assert.Equal(Path.Combine(ctx.StreamsRoot, item.ToString()), Path.GetDirectoryName(StreamPath(ctx, item, file.FileName))); + var lines = await File.ReadAllLinesAsync(StreamPath(ctx, item, file.FileName), cts.Token); + Assert.Equal(["alpha", "beta"], lines); + + // The equivalent local phase keys its artefact identically: the same + // work-item directory layout and the same phase/iteration file prefix. + var twin = WorkItemId.New(); + await using (var local = await ctx.Streams.BeginCaptureAsync(twin, "work", 1, cts.Token)) + { + Assert.NotNull(local); + local!.WriteChunk("alpha\nbeta\n"); + } + + var twinFile = Assert.Single(await ctx.Streams.ListAsync(twin, ct: cts.Token)); + Assert.Equal(Path.Combine(ctx.StreamsRoot, twin.ToString()), Path.GetDirectoryName(StreamPath(ctx, twin, twinFile.FileName))); + Assert.StartsWith("work-1-", twinFile.FileName); + var twinLines = await File.ReadAllLinesAsync(StreamPath(ctx, twin, twinFile.FileName), cts.Token); + Assert.Equal(lines, twinLines); + + // Live subscribers saw the same chunks through the existing contract. + Assert.Equal(["work", "work"], ctx.Broadcaster.Phases); + Assert.Contains("alpha\n", ctx.Broadcaster.Text); + Assert.Contains("beta\n", ctx.Broadcaster.Text); + } + + // ── verification 2: live delivery during execution ────────────────────── + + [Fact] + public async Task LiveSubscriber_ReceivesChunks_DuringExecution_NotOnlyAtCompletion() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + using var ctx = CreateContext(); + var item = WorkItemId.New(); + await ctx.Git.EnsureRepositoryAsync(item, seedFromUrl: null, cts.Token); + var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + ctx.Transport.Script = _ => [(0, "live-one\n")]; + ctx.Transport.BeforeReturn = async token => + { + await gate.Task.WaitAsync(token).ConfigureAwait(false); + await Task.Delay(50, token).ConfigureAwait(false); + }; + + var dispatch = ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0), cts.Token); + await ctx.Broadcaster.FirstChunk.Task.WaitAsync(TimeSpan.FromSeconds(10), cts.Token); + + // The first chunk arrived while the phase was still running: the + // transport is parked behind the test gate, so completion is + // impossible yet. + Assert.False(dispatch.IsCompleted); + gate.TrySetResult(); + var result = await dispatch.WaitAsync(TimeSpan.FromSeconds(20), cts.Token); + + Assert.Equal(ExecutorPhaseOutcome.Succeeded, result.Outcome); + var file = Assert.Single(await ctx.Streams.ListAsync(item, ct: cts.Token)); + Assert.Contains("live-one", await File.ReadAllTextAsync(StreamPath(ctx, item, file.FileName), cts.Token)); + } + + // ── verification 3: over-limit truncation marker parity ───────────────── + + [Fact] + public async Task RemoteProducer_BeyondBufferingLimits_IsTruncatedWithSameMarkerAsLocal() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(120)); + using var ctx = CreateContext(maxFileSizeMb: 1); + var item = WorkItemId.New(); + await ctx.Git.EnsureRepositoryAsync(item, seedFromUrl: null, cts.Token); + + var script = new List<(long, string)>(); + for (var i = 0; i < 12; i++) + script.Add((i, BuildBlock(1000))); + ctx.Transport.Script = _ => script; + + var result = await ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0), cts.Token); + + Assert.Equal(ExecutorPhaseOutcome.Succeeded, result.Outcome); + var file = Assert.Single(await ctx.Streams.ListAsync(item, ct: cts.Token)); + var path = StreamPath(ctx, item, file.FileName); + Assert.True(new FileInfo(path).Length <= 1024 * 1024); + var remoteLines = await File.ReadAllLinesAsync(path, cts.Token); + var remoteMarker = Assert.Single(remoteLines, l => l.StartsWith("[...truncated by ", StringComparison.Ordinal)); + + var twin = WorkItemId.New(); + await using (var local = await ctx.Streams.BeginCaptureAsync(twin, "work", 1, cts.Token)) + { + Assert.NotNull(local); + for (var i = 0; i < 12; i++) + local!.WriteChunk(BuildBlock(1000)); + } + + var twinFile = Assert.Single(await ctx.Streams.ListAsync(twin, ct: cts.Token)); + var twinLines = await File.ReadAllLinesAsync(StreamPath(ctx, twin, twinFile.FileName), cts.Token); + var localMarker = Assert.Single(twinLines, l => l.StartsWith("[...truncated by ", StringComparison.Ordinal)); + Assert.Equal(localMarker, remoteMarker); + } + + // ── verification 4: relay failure is observability-only ───────────────── + + [Fact] + public async Task RelayBroadcastFailure_LeavesPhaseOutcomeUnchanged_AndRetainsPartialStream() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + using var ctx = CreateContext(); + ctx.Broadcaster.ShouldThrow = _ => true; + var item = WorkItemId.New(); + await ctx.Git.EnsureRepositoryAsync(item, seedFromUrl: null, cts.Token); + ctx.Transport.Script = _ => [(0, "kept-one\n"), (1, "kept-two\n")]; + + var result = await ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0), cts.Token); + + Assert.Equal(ExecutorPhaseOutcome.Succeeded, result.Outcome); + var file = Assert.Single(await ctx.Streams.ListAsync(item, ct: cts.Token)); + var lines = await File.ReadAllLinesAsync(StreamPath(ctx, item, file.FileName), cts.Token); + Assert.Equal(["kept-one", "kept-two"], lines); + } + + [Fact] + public async Task RelayCaptureFailure_LeavesPhaseOutcomeUnchanged_AndStillBroadcastsLive() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + using var ctx = CreateContext(streamStore: new ThrowingStreamStore()); + var item = WorkItemId.New(); + await ctx.Git.EnsureRepositoryAsync(item, seedFromUrl: null, cts.Token); + ctx.Transport.Script = _ => [(0, "live-despite-capture-failure\n")]; + + var result = await ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0), cts.Token); + + Assert.Equal(ExecutorPhaseOutcome.Succeeded, result.Outcome); + Assert.Contains("live-despite-capture-failure\n", ctx.Broadcaster.Text); + } + + // ── verification 5: gaps are recorded, never silent ───────────────────── + + [Fact] + public async Task InducedGapInRelayedSequence_IsRecordedInCapturedStream() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + using var ctx = CreateContext(); + var item = WorkItemId.New(); + await ctx.Git.EnsureRepositoryAsync(item, seedFromUrl: null, cts.Token); + ctx.Transport.Script = _ => [(0, "first\n"), (2, "third\n")]; + + var result = await ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0), cts.Token); + + Assert.Equal(ExecutorPhaseOutcome.Succeeded, result.Outcome); + var file = Assert.Single(await ctx.Streams.ListAsync(item, ct: cts.Token)); + var lines = await File.ReadAllLinesAsync(StreamPath(ctx, item, file.FileName), cts.Token); + Assert.Equal(3, lines.Length); + Assert.Equal("first", lines[0]); + Assert.Contains("stream gap", lines[1], StringComparison.Ordinal); + Assert.Contains("expected seq 1", lines[1], StringComparison.Ordinal); + Assert.Contains("received seq 2", lines[1], StringComparison.Ordinal); + Assert.Equal("third", lines[2]); + Assert.Contains("stream gap", ctx.Broadcaster.Text, StringComparison.Ordinal); + } + + private static string BuildBlock(int lines) + { + var sb = new StringBuilder(lines * 101); + for (var i = 0; i < lines; i++) + sb.Append('x', 100).Append('\n'); + return sb.ToString(); + } + + private static string StreamPath(RelayContext ctx, WorkItemId item, string fileName) => + Path.Combine(ctx.StreamsRoot, item.ToString(), fileName); + + private static ExecutorPhaseRequest NewRequest(WorkItemId item, string phase, int attempt) => + new() { WorkItemId = item.ToString(), Phase = phase, Attempt = attempt, RepositoryId = item.ToString(), PayloadJson = "{}" }; + + private RelayContext CreateContext(int maxFileSizeMb = 32, IAgentStreamStore? streamStore = null) + { + var gitRoot = Path.Combine(_root, "git-" + Guid.NewGuid().ToString("N")); + var git = new LocalGitHost( + new LocalGitHostOptions { RootDirectory = gitRoot }, + NullLogger.Instance); + var dbPath = Path.Combine(_root, "idem-" + Guid.NewGuid().ToString("N") + ".db"); + var idempotency = new SqliteIdempotencyStore(dbPath); + var registry = new FakeWorkerRegistry(); + var options = new ExecutorPhaseDispatchOptions(); + var transport = new FakeStreamingTransport("exec-1", Path.Combine(_root, "executor-" + Guid.NewGuid().ToString("N"))); + var factory = new FakeTransportFactory(transport); + var inner = new InProcessExecutorPhaseRunner(git, new UncalledHandler(), () => options); + var streamsRoot = Path.Combine(_root, "streams-" + Guid.NewGuid().ToString("N")); + var streams = new AgentStreamStore( + new AgentStreamsOptions { Enabled = true, Path = streamsRoot, MaxFileSizeMb = maxFileSizeMb }, + NullLogger.Instance); + var broadcaster = new RecordingBroadcaster(); + registry.AddExecutor("exec-1"); + var proxy = new ExecutorPhaseProxy( + registry, factory, git, idempotency, inner, () => options, + streamStore: streamStore ?? streams, broadcaster: broadcaster); + return new RelayContext(git, idempotency, options, transport, streams, streamsRoot, broadcaster, proxy); + } + + private sealed record RelayContext( + LocalGitHost Git, + SqliteIdempotencyStore Idempotency, + ExecutorPhaseDispatchOptions Options, + FakeStreamingTransport Transport, + AgentStreamStore Streams, + string StreamsRoot, + RecordingBroadcaster Broadcaster, + ExecutorPhaseProxy Proxy) : IDisposable + { + public void Dispose() => Idempotency.Dispose(); + } + + private sealed class UncalledHandler : IExecutorPhaseHandler + { + public Task ExecuteAsync(ExecutorPhaseRequest request, string repoPath, CancellationToken ct) => + throw new InvalidOperationException("Fallback runner must not run while an executor is registered."); + } + + private sealed class RecordingBroadcaster : IStdoutBroadcaster + { + private readonly object _lock = new(); + private readonly List<(string Phase, string Chunk)> _chunks = []; + + public TaskCompletionSource FirstChunk { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public Func? ShouldThrow; + + public IReadOnlyList Phases + { + get { lock (_lock) { return [.. _chunks.Select(c => c.Phase)]; } } + } + + public string Text + { + get { lock (_lock) { return string.Concat(_chunks.Select(c => c.Chunk)); } } + } + + public void BroadcastChunk(WorkItemId workItemId, string phase, string chunk) + { + lock (_lock) + { + if (ShouldThrow?.Invoke(_chunks.Count) == true) + throw new InvalidOperationException("Simulated broadcast failure."); + _chunks.Add((phase, chunk)); + } + FirstChunk.TrySetResult(); + } + + public Task CompleteAsync(WorkItemId workItemId) => Task.CompletedTask; + + public string? GetTail(WorkItemId workItemId) + { + lock (_lock) + return _chunks.Count == 0 ? null : string.Concat(_chunks.Select(c => c.Chunk)); + } + } + + private sealed class ThrowingStreamStore : IAgentStreamStore + { + public AgentStreamsOptions Options => new() { Enabled = true, Path = Path.GetTempPath() }; + + public Task BeginCaptureAsync(WorkItemId workItemId, string phase, int iteration, CancellationToken ct = default) => + throw new InvalidOperationException("Simulated stream store failure."); + + public Task> ListAsync(WorkItemId workItemId, int limit = 100, bool includeLineCount = false, CancellationToken ct = default) => + throw new NotSupportedException(); + + public Task GetAsync(WorkItemId workItemId, string fileName, bool includeLineCount = false, CancellationToken ct = default) => + throw new NotSupportedException(); + + public Task OpenReadAsync(WorkItemId workItemId, string fileName, CancellationToken ct = default) => + throw new NotSupportedException(); + + public Task SweepAsync(DateTimeOffset now, CancellationToken ct = default) => + throw new NotSupportedException(); + } + + private sealed class FakeWorkerRegistry : IWorkerRegistry + { + private readonly Dictionary _rows = new(StringComparer.Ordinal); + + public void AddExecutor(string hostId) + { + var now = DateTimeOffset.UtcNow; + _rows[ExecutorRegistration.WorkerIdFor(hostId)] = new WorkerRegistration + { + WorkerId = ExecutorRegistration.WorkerIdFor(hostId), + HostName = hostId, + ProcessId = 4242, + StartedAt = now, + LastHeartbeatAt = now, + ExecutorHostId = hostId, + MaxConcurrentSandboxes = 4, + ExecutorNetworkProfiles = [], + ExecutorCredentials = [], + Cordoned = false, + Healthy = true, + }; + } + + public Task RegisterAsync(WorkerRegistration registration, CancellationToken ct = default) + { + _rows[registration.WorkerId] = registration; + return Task.CompletedTask; + } + + public Task HeartbeatAsync(string workerId, string? currentWorkItemId, CancellationToken ct = default) + { + if (_rows.TryGetValue(workerId, out var row)) + _rows[workerId] = row with { LastHeartbeatAt = DateTimeOffset.UtcNow, CurrentWorkItemId = currentWorkItemId }; + return Task.CompletedTask; + } + + public Task DeregisterAsync(string workerId, CancellationToken ct = default) + { + _rows.Remove(workerId); + return Task.CompletedTask; + } + + public Task> ListAsync(CancellationToken ct = default) + { + IReadOnlyList snapshot = [.. _rows.Values]; + return Task.FromResult(snapshot); + } + + public Task> ClaimDeadWorkersAsync(DateTimeOffset cutoff, CancellationToken ct = default) => + Task.FromResult>([]); + + public Task TryClaimDeadWorkerAsync(string workerId, DateTimeOffset cutoff, CancellationToken ct = default) => + Task.FromResult(null); + + public Task TryClaimWorkerAsync(string workerId, CancellationToken ct = default) => + Task.FromResult(null); + } + + private sealed class FakeTransportFactory : IExecutorPhaseTransportFactory + { + private readonly FakeStreamingTransport _transport; + + public FakeTransportFactory(FakeStreamingTransport transport) => _transport = transport; + + public Task ResolveAsync(string hostId, CancellationToken ct) => + Task.FromResult(_transport); + } + + /// + /// Loopback executor transport with a scripted sequenced chunk stream: + /// stage-in copies the single repo path over, the phase emits its script + /// through the streaming callback before returning a canned success, and + /// stage-out tars the copy back through real tar bytes. + /// + private sealed class FakeStreamingTransport : IStreamingExecutorPhaseTransport + { + public FakeStreamingTransport(string hostId, string executorRoot) + { + HostId = hostId; + ExecutorRoot = executorRoot; + Directory.CreateDirectory(executorRoot); + } + + public string HostId { get; } + public string ExecutorRoot { get; } + public Func>? Script; + public Func? BeforeReturn; + public int RunPhaseCalls { get; private set; } + + public string StagedCopy + { + get + { + var entries = Directory.GetFileSystemEntries(ExecutorRoot); + return entries.Length == 1 ? entries[0] : throw new InvalidOperationException("No staged repo on fake executor."); + } + } + + public Task StageInAsync(string hostRepoPath, CancellationToken ct) + { + var dest = Path.Combine(ExecutorRoot, Path.GetFileName(hostRepoPath.TrimEnd(Path.DirectorySeparatorChar))); + CopyDirectory(hostRepoPath, dest); + return Task.CompletedTask; + } + + public Task RunPhaseAsync(ExecutorPhaseRequest request, CancellationToken ct) => + RunPhaseAsync(request, onChunk: null, ct); + + public async Task RunPhaseAsync( + ExecutorPhaseRequest request, + Func? onChunk, + CancellationToken ct) + { + RunPhaseCalls++; + if (Script is not null) + { + foreach (var (sequence, data) in Script(request)) + { + ct.ThrowIfCancellationRequested(); + if (onChunk is not null) + await onChunk(new ExecutorStreamChunk { Sequence = sequence, Data = data }, ct).ConfigureAwait(false); + } + } + if (BeforeReturn is not null) + await BeforeReturn(ct).ConfigureAwait(false); + return new ExecutorPhaseResult + { + Outcome = ExecutorPhaseOutcome.Succeeded, + Findings = [$"finding for {request.Phase}"], + Usage = new ExecutorPhaseUsage(100, 50, 0.002m), + }; + } + + public async Task StageOutToArchiveAsync(string hostArchivePath, long maxArchiveBytes, CancellationToken ct) + { + await using var file = File.OpenWrite(hostArchivePath); + await WriteTarOfDirectoryAsync(StagedCopy, file, ct).ConfigureAwait(false); + } + + private static void CopyDirectory(string source, string destination) + { + if (Directory.Exists(destination)) + Directory.Delete(destination, recursive: true); + Directory.CreateDirectory(destination); + foreach (var dir in Directory.GetDirectories(source, "*", SearchOption.AllDirectories)) + Directory.CreateDirectory(Path.Combine(destination, Path.GetRelativePath(source, dir))); + foreach (var file in Directory.GetFiles(source, "*", SearchOption.AllDirectories)) + File.Copy(file, Path.Combine(destination, Path.GetRelativePath(source, file)), overwrite: true); + } + + private static async Task WriteTarOfDirectoryAsync(string sourceDir, Stream destination, CancellationToken ct) + { + var rootName = Path.GetFileName(sourceDir.TrimEnd(Path.DirectorySeparatorChar)); + await using var writer = new TarWriter(destination, TarEntryFormat.Pax, leaveOpen: true); + foreach (var dir in Directory.GetDirectories(sourceDir, "*", SearchOption.AllDirectories).OrderBy(x => x, StringComparer.Ordinal)) + { + ct.ThrowIfCancellationRequested(); + var name = rootName + "/" + Path.GetRelativePath(sourceDir, dir).Replace('\\', '/'); + await writer.WriteEntryAsync(new PaxTarEntry(TarEntryType.Directory, name), ct).ConfigureAwait(false); + } + foreach (var file in Directory.GetFiles(sourceDir, "*", SearchOption.AllDirectories).OrderBy(x => x, StringComparer.Ordinal)) + { + ct.ThrowIfCancellationRequested(); + var name = rootName + "/" + Path.GetRelativePath(sourceDir, file).Replace('\\', '/'); + var entry = new PaxTarEntry(TarEntryType.RegularFile, name); + await using var data = File.OpenRead(file); + entry.DataStream = data; + await writer.WriteEntryAsync(entry, ct).ConfigureAwait(false); + } + } + } +}