diff --git a/docs/operating/remote-executors.md b/docs/operating/remote-executors.md index 01578e89..5071f47b 100644 --- a/docs/operating/remote-executors.md +++ b/docs/operating/remote-executors.md @@ -7,9 +7,43 @@ phase-execution seam. It connects **outbound** to the orchestrator (plain HTTPS POSTs) and never opens an inbound listening port, so it can sit behind NAT or a host firewall. -This page covers the process, its registration and its liveness. Dispatching -work to a registered executor is a separate item: an executor that registers -and heartbeats but is never sent work is the acceptance state. +This page covers the process, its registration, its liveness, and the +phase-dispatch proxy that sends work to a registered executor. + +## Dispatching phases to an executor + +`ExecutorPhaseProxy` (`src/CodeyBox.Orchestrator/ExecutorPhaseProxy.cs`) +implements `IExecutorPhaseRunner`: it selects a registered executor from the +worker registry using `ExecutorEligibility` (zero-capacity, cordoned and +unhealthy hosts register but are never selected), stages the phase's single +bare repo to the host through `IExecutorPhaseTransport`, runs the phase +there, and stages the repo back as a tar archive that is validated (archive +bytes, entry count, expansion ratio, path containment) before anything is +extracted over the orchestrator's bare repo. The archive-byte cap is enforced +by the transport while receiving — an unbounded payload is aborted mid-stream +rather than buffered to disk and rejected afterwards — with the validator +re-checking the landed size as defense in depth. Only the per-item repo is ever +transferred — never the whole repos root — so an executor receives only the +repo for the item it is running. + +Delivery is idempotent through `IIdempotencyStore`: the key is work item + +phase + attempt and the body hash covers the request, so a redelivered +dispatch replays the original result instead of provisioning a second +sandbox, while the same key with a different body is refused as a conflict +and never executes. With no executor registered, dispatch falls back to the +in-process runner with unchanged behaviour. + +An agent failure on the executor is returned as a result (`AgentFailed`); a +host, connection or transfer problem throws `ExecutorPhaseTransportException` +and stores nothing, so an unreachable host is retried elsewhere rather than +charged against the work item as an agent failure. The proxy never touches +the work item table — the transport carries dispatch only, and re-dispatch +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. ## Running the executor diff --git a/src/CodeyBox.Core/ExecutorPhase.cs b/src/CodeyBox.Core/ExecutorPhase.cs new file mode 100644 index 00000000..5090574e --- /dev/null +++ b/src/CodeyBox.Core/ExecutorPhase.cs @@ -0,0 +1,88 @@ +using System.Text.Json.Serialization; + +namespace CodeyBox.Core; + +/// +/// Outcome of a phase executed on an executor host (or in process). +/// A phase that fails because the host was unreachable or the transfer broke +/// is NOT represented here — that surfaces as +/// so the caller retries +/// elsewhere instead of charging the work item with an agent failure. +/// +public enum ExecutorPhaseOutcome +{ + Succeeded = 0, + AgentFailed = 1, +} + +/// +/// Resource usage attributed to one phase execution. All values are produced +/// by the code under test (the executor-side handler or its in-process +/// twin) and are validated non-negative by the dispatch proxy before they +/// are cached or returned. +/// +public sealed record ExecutorPhaseUsage( + [property: JsonPropertyName("inputTokens")] long InputTokens, + [property: JsonPropertyName("outputTokens")] long OutputTokens, + [property: JsonPropertyName("costUsd")] decimal CostUsd); + +/// +/// Dispatch envelope for one phase of a work item. The proxy keys idempotent +/// delivery on work item + phase + attempt: a redelivered dispatch (same key, +/// same body hash) returns the original result instead of provisioning a +/// second sandbox, while a new attempt uses a new key and executes fresh. +/// +public sealed record ExecutorPhaseRequest +{ + /// Work item the phase belongs to. Non-empty, at most 128 chars. + public required string WorkItemId { get; init; } + + /// + /// Phase name (for example "work", "audit", "merge"). Open vocabulary so + /// future pipeline phases need no contract change; restricted to + /// [A-Za-z0-9_-], at most 64 chars. + /// + public required string Phase { get; init; } + + /// + /// Attempt number within the phase. Must be zero or positive; redelivery + /// of the same attempt is idempotent, a new attempt is a new dispatch. + /// + public required int Attempt { get; init; } + + /// + /// Bare-repo id to stage to the executor (normally the work item id). + /// Only this repo is transferred — never the whole repos root. + /// + public required string RepositoryId { get; init; } + + /// + /// Serialized phase input (for example the work item snapshot). Bounded + /// by dispatch options; covered by the idempotency body hash. + /// + public required string PayloadJson { get; init; } +} + +/// +/// Result of one phase execution: agent-visible outcome plus the commit the +/// phase produced, the findings it reported, and the usage it consumed. +/// +public sealed record ExecutorPhaseResult +{ + public required ExecutorPhaseOutcome Outcome { get; init; } + + /// + /// Full hex commit sha the phase left on its branch, if it produced one. + /// Lowercase hex, 40 (SHA-1) or 64 (SHA-256) chars, or null/empty when + /// the phase produced no commit. + /// + public string? CommitSha { get; init; } + + /// Findings reported by the phase (for example audit findings). + public IReadOnlyList Findings { get; init; } = []; + + public required ExecutorPhaseUsage Usage { get; init; } + + /// Agent-facing error detail for . + public string? ErrorMessage { get; init; } +} diff --git a/src/CodeyBox.Core/ExecutorPhaseTransport.cs b/src/CodeyBox.Core/ExecutorPhaseTransport.cs new file mode 100644 index 00000000..1303c39f --- /dev/null +++ b/src/CodeyBox.Core/ExecutorPhaseTransport.cs @@ -0,0 +1,131 @@ +namespace CodeyBox.Core; + +/// +/// Transport failure moving a phase to or from an executor host: the host was +/// unreachable, authentication failed, the connection dropped mid-transfer, +/// or no transport is configured for the selected host. The phase itself did +/// not fail — the caller must retry elsewhere (another host or the existing +/// pipeline recovery path) without charging the work item with an agent +/// failure or consuming a rework iteration. Carries host and operation only. +/// +public sealed class ExecutorPhaseTransportException : Exception +{ + public ExecutorPhaseTransportException(string hostId, string operation, string message) + : base($"Executor phase {operation} failed on host '{hostId}': {message}") + { + HostId = hostId; + Operation = operation; + } + + public ExecutorPhaseTransportException(string hostId, string operation, string message, Exception inner) + : base($"Executor phase {operation} failed on host '{hostId}': {message}", inner) + { + HostId = hostId; + Operation = operation; + } + + public string HostId { get; } + public string Operation { get; } +} + +/// +/// The phase ran (or was answered) but cannot be accepted: the staged-back +/// payload exceeded its bounds, failed validation, or the executor's result +/// was malformed. Distinct from : +/// the host was reachable, so retrying on another host with the same payload +/// shape may legitimately fail the same way. The orchestrator's bare repo is +/// never written when this is thrown, and no idempotency record is stored so +/// a corrected redelivery can still execute. +/// +public sealed class ExecutorPhaseException : Exception +{ + public ExecutorPhaseException(string message) + : base(message) + { + } + + public ExecutorPhaseException(string message, Exception inner) + : base(message, inner) + { + } +} + +/// +/// Idempotency conflict: the dispatch key (work item + phase + attempt) +/// already delivered a result for a DIFFERENT request body. The redelivery +/// must not execute — something is dispatching inconsistent inputs under one +/// attempt. Carries the key only, never request bodies. +/// +public sealed class ExecutorPhaseConflictException : Exception +{ + public ExecutorPhaseConflictException(string dispatchKey) + : base($"Executor phase dispatch key '{dispatchKey}' was already used with a different request body; refusing to execute.") + { + DispatchKey = dispatchKey; + } + + public string DispatchKey { get; } +} + +/// +/// Dispatch-only channel to one executor host. Carries the phase request and +/// the per-item bare repo to the host and back; it persists no queue state. +/// The work item table remains the queue of record — re-dispatch after +/// failure is driven by the existing pipeline state machine and recovery +/// paths, which call back into the dispatch proxy with a new attempt. +/// +/// All methods throw on +/// transport failure. Implementations must stage exactly the repo path they +/// are given — never the whole repos root — so an executor receives only the +/// repo for the item it is running. Exception messages must carry host and +/// operation only: never key material, request bodies, or raw remote output. +/// +public interface IExecutorPhaseTransport +{ + /// Stable executor host id this transport talks to. + string HostId { get; } + + /// + /// Copies the host-local bare repo at to + /// the executor. Called with exactly one per-item repo path per dispatch. + /// + Task StageInAsync(string hostRepoPath, CancellationToken ct); + + /// + /// Runs the phase on the executor against its staged repo copy and + /// returns the phase result. An agent failure on the executor is returned + /// as a result with , not + /// thrown — only transport failures throw. + /// + Task RunPhaseAsync(ExecutorPhaseRequest request, CancellationToken ct); + + /// + /// Writes the executor-side repo back to a host-local tar archive at + /// . The caller validates the archive + /// (size, entry count, expansion ratio, path containment) before anything + /// is extracted over the orchestrator's bare repo. + /// + /// The transport MUST enforce + /// while receiving — aborting the transfer as soon as the cap is + /// exceeded — rather than buffering an unbounded payload and reporting + /// its size afterwards. The executor is untrusted, so a post-write size + /// check alone lets a compromised executor fill the orchestrator disk + /// before validation rejects the payload. Exceeding the cap throws + /// (a phase failure: the host was + /// reachable, the payload was hostile), never a transport exception, and + /// must leave no usable archive behind at + /// . + /// + Task StageOutToArchiveAsync(string hostArchivePath, long maxArchiveBytes, 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 +/// is mapped for it); the proxy treats that as a transport failure rather +/// than silently running the phase elsewhere. +/// +public interface IExecutorPhaseTransportFactory +{ + Task ResolveAsync(string hostId, CancellationToken ct); +} diff --git a/src/CodeyBox.Orchestrator/ExecutorPhaseDispatchOptions.cs b/src/CodeyBox.Orchestrator/ExecutorPhaseDispatchOptions.cs new file mode 100644 index 00000000..8d9051fc --- /dev/null +++ b/src/CodeyBox.Orchestrator/ExecutorPhaseDispatchOptions.cs @@ -0,0 +1,84 @@ +namespace CodeyBox.Orchestrator; + +/// +/// Tuning knobs for executor phase dispatch (see ). +/// Bound under CodeyBox:ExecutorPhaseDispatch. The whole record is +/// hot-reloadable through a delegate accessor — a config edit lands on the +/// next dispatch without an orchestrator restart. Operational values live +/// here, never as literals in the proxy or validator. +/// +public sealed class ExecutorPhaseDispatchOptions +{ + /// + /// Maximum tar bytes accepted back from an executor per dispatch. The + /// transport enforces this cap while receiving (see + /// IExecutorPhaseTransport.StageOutToArchiveAsync) so + /// executor-controlled content cannot fill the orchestrator disk before + /// validation; the validator re-checks the landed size as defense in + /// depth. + /// Equivalent to MultipassRemoteSandboxOptions.StageOutMaxArchiveBytes. + /// + public long StageOutMaxArchiveBytes { get; set; } = 2L * 1024 * 1024 * 1024; + + /// + /// Maximum non-metadata tar entries accepted per staged-back archive. + /// Equivalent to MultipassRemoteSandboxOptions.StageOutMaxEntries. + /// + public int StageOutMaxEntries { get; set; } = 200_000; + + /// + /// Maximum declared regular-file payload divided by archive bytes. + /// Equivalent to MultipassRemoteSandboxOptions.StageOutMaxExpansionRatio. + /// + public double StageOutMaxExpansionRatio { get; set; } = 1.5d; + + /// + /// How long a delivered dispatch result is replayed from the idempotency + /// store on redelivery. Mirrors the API idempotency TTL. + /// + public TimeSpan IdempotencyTtl { get; set; } = TimeSpan.FromHours(24); + + /// Maximum serialized bytes accepted in a dispatch request payload. + public int MaxRequestPayloadBytes { get; set; } = 1024 * 1024; + + /// Maximum findings accepted in an executor-returned result. + public int MaxResultFindings { get; set; } = 128; + + /// Maximum chars accepted per finding in an executor-returned result. + public int MaxFindingLengthChars { get; set; } = 8192; + + /// Maximum chars accepted in an executor-returned error message. + public int MaxResultErrorLengthChars { get; set; } = 8192; + + /// + /// Fails fast on misconfiguration so a bad bound surfaces at dispatch + /// time instead of silently admitting an unbounded payload. + /// + public void Validate() + { + if (StageOutMaxArchiveBytes <= 0) + throw new InvalidOperationException( + "CodeyBox:ExecutorPhaseDispatch:StageOutMaxArchiveBytes must be > 0."); + if (StageOutMaxEntries <= 0) + throw new InvalidOperationException( + "CodeyBox:ExecutorPhaseDispatch:StageOutMaxEntries must be > 0."); + if (double.IsNaN(StageOutMaxExpansionRatio) || double.IsInfinity(StageOutMaxExpansionRatio) || StageOutMaxExpansionRatio < 1.0) + throw new InvalidOperationException( + "CodeyBox:ExecutorPhaseDispatch:StageOutMaxExpansionRatio must be a finite value >= 1.0."); + if (IdempotencyTtl <= TimeSpan.Zero) + throw new InvalidOperationException( + "CodeyBox:ExecutorPhaseDispatch:IdempotencyTtl must be positive."); + if (MaxRequestPayloadBytes <= 0) + throw new InvalidOperationException( + "CodeyBox:ExecutorPhaseDispatch:MaxRequestPayloadBytes must be > 0."); + if (MaxResultFindings <= 0) + throw new InvalidOperationException( + "CodeyBox:ExecutorPhaseDispatch:MaxResultFindings must be > 0."); + if (MaxFindingLengthChars <= 0) + throw new InvalidOperationException( + "CodeyBox:ExecutorPhaseDispatch:MaxFindingLengthChars must be > 0."); + if (MaxResultErrorLengthChars <= 0) + throw new InvalidOperationException( + "CodeyBox:ExecutorPhaseDispatch:MaxResultErrorLengthChars must be > 0."); + } +} diff --git a/src/CodeyBox.Orchestrator/ExecutorPhaseProxy.cs b/src/CodeyBox.Orchestrator/ExecutorPhaseProxy.cs new file mode 100644 index 00000000..0cb5d11b --- /dev/null +++ b/src/CodeyBox.Orchestrator/ExecutorPhaseProxy.cs @@ -0,0 +1,381 @@ +using System.Security.Cryptography; +using System.Text; +using System.Text.Json; +using System.Text.RegularExpressions; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using CodeyBox.Core; + +namespace CodeyBox.Orchestrator; + +/// +/// Proxy that dispatches a phase to a +/// registered executor host and makes delivery safe to retry. +/// +/// Per dispatch, in order: +/// +/// Validate the request bounds and compute the idempotency key (work +/// item + phase + attempt) plus the SHA-256 body hash. +/// Look the key up in . A Hit replays the +/// original result without provisioning a second sandbox; a Conflict (same +/// key, different body) throws +/// and never executes. +/// Select a registered executor from +/// using (zero-capacity, cordoned and +/// unhealthy hosts register but are never selected). With no executor +/// registered, fall back to the in-process runner with unchanged behaviour. +/// 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. +/// 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. +/// Cache the result under the dispatch key and return it. +/// +/// +/// Failure taxonomy: an agent failure on the executor is returned as a +/// result with ; a host, +/// connection or transfer problem throws +/// and stores nothing, so an +/// unreachable host is retried elsewhere rather than charged against the +/// work item as an agent failure. The proxy never touches the work item +/// table — it carries dispatch only; re-dispatch after failure is driven by +/// the existing pipeline state machine with a new attempt. +/// +public sealed class ExecutorPhaseProxy : IExecutorPhaseRunner +{ + private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web); + private static readonly Regex PhaseNamePattern = new("^[A-Za-z0-9_-]{1,64}$", RegexOptions.CultureInvariant | RegexOptions.Compiled); + + private const int MaxWorkItemIdLength = 128; + private const int MaxRepositoryIdLength = 256; + + private readonly IWorkerRegistry _registry; + private readonly IExecutorPhaseTransportFactory _transports; + private readonly IGitHost _gitHost; + private readonly IIdempotencyStore _idempotency; + private readonly IExecutorPhaseRunner _inner; + private readonly Func _optionsAccessor; + private readonly TimeProvider _clock; + private readonly ILogger _log; + + public ExecutorPhaseProxy( + IWorkerRegistry registry, + IExecutorPhaseTransportFactory transports, + IGitHost gitHost, + IIdempotencyStore idempotency, + IExecutorPhaseRunner inner, + Func optionsAccessor, + TimeProvider? clock = null, + ILogger? log = null) + { + _registry = registry ?? throw new ArgumentNullException(nameof(registry)); + _transports = transports ?? throw new ArgumentNullException(nameof(transports)); + _gitHost = gitHost ?? throw new ArgumentNullException(nameof(gitHost)); + _idempotency = idempotency ?? throw new ArgumentNullException(nameof(idempotency)); + _inner = inner ?? throw new ArgumentNullException(nameof(inner)); + _optionsAccessor = optionsAccessor ?? throw new ArgumentNullException(nameof(optionsAccessor)); + _clock = clock ?? TimeProvider.System; + _log = log ?? NullLogger.Instance; + } + + public async Task ExecutePhaseAsync(ExecutorPhaseRequest request, CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(request); + var options = _optionsAccessor(); + options.Validate(); + ValidateRequest(request, options); + + var dispatchKey = BuildDispatchKey(request); + var bodyHash = ComputeBodyHash(request); + var now = _clock.GetUtcNow(); + + var lookup = await _idempotency.LookupAsync(dispatchKey, bodyHash, now, ct).ConfigureAwait(false); + switch (lookup.Outcome) + { + case IdempotencyLookupOutcome.Hit: + _log.LogInformation("Executor phase dispatch {DispatchKey} redelivered; replaying original result", dispatchKey); + return ValidateResult(DeserializeCachedResult(dispatchKey, lookup.Entry!), options); + case IdempotencyLookupOutcome.Conflict: + throw new ExecutorPhaseConflictException(dispatchKey); + case IdempotencyLookupOutcome.Miss: + break; + default: + throw new InvalidOperationException($"Unknown idempotency outcome {(int)lookup.Outcome}."); + } + + var hostId = await SelectExecutorAsync(ct).ConfigureAwait(false); + if (hostId is null) + { + _log.LogInformation("No executor registered for dispatch {DispatchKey}; falling back to in-process execution", dispatchKey); + var fallback = await _inner.ExecutePhaseAsync(request, ct).ConfigureAwait(false); + var validatedFallback = ValidateResult(fallback, options); + await _idempotency.PutAsync( + new IdempotencyEntry(dispatchKey, bodyHash, 200, SerializeResult(validatedFallback), "application/json", now + options.IdempotencyTtl), + ct).ConfigureAwait(false); + return validatedFallback; + } + + var result = await ExecuteRemoteAsync(request, hostId, options, ct).ConfigureAwait(false); + await _idempotency.PutAsync( + new IdempotencyEntry(dispatchKey, bodyHash, 200, SerializeResult(result), "application/json", now + options.IdempotencyTtl), + ct).ConfigureAwait(false); + return result; + } + + private async Task ExecuteRemoteAsync( + ExecutorPhaseRequest request, + string hostId, + ExecutorPhaseDispatchOptions options, + CancellationToken ct) + { + var transport = await ResolveTransportAsync(hostId, ct).ConfigureAwait(false); + + string repoPath; + try + { + repoPath = _gitHost.GetRepoPath(request.RepositoryId); + } + catch (NotSupportedException ex) + { + throw new ExecutorPhaseTransportException(hostId, "resolve-repo", "Git host exposes no local repository path.", ex); + } + var canonicalRepo = CanonicalizeRepoPath(repoPath, _gitHost.RepositoriesRootDirectory, request.RepositoryId); + if (!Directory.Exists(canonicalRepo)) + throw new ExecutorPhaseException($"Bare repo for '{request.RepositoryId}' does not exist."); + + 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); + var result = ValidateResult(raw, options); + + var scratchRoot = Path.Combine(Path.GetTempPath(), "codeybox-executor-phase-" + Guid.NewGuid().ToString("N")); + Directory.CreateDirectory(scratchRoot); + try + { + var archivePath = Path.Combine(scratchRoot, "stageout.tar"); + await CallTransportAsync(hostId, "stage-out", token => transport.StageOutToArchiveAsync(archivePath, options.StageOutMaxArchiveBytes, token), ct).ConfigureAwait(false); + await ExecutorStageOutValidator.ValidateAndInstallAsync(archivePath, canonicalRepo, scratchRoot, options, ct).ConfigureAwait(false); + } + finally + { + try { if (Directory.Exists(scratchRoot)) Directory.Delete(scratchRoot, recursive: true); } + catch { } + } + + _log.LogInformation("Executor phase dispatch for host {HostId} completed with outcome {Outcome}", hostId, result.Outcome); + return result; + } + + private async Task ResolveTransportAsync(string hostId, CancellationToken ct) + { + IExecutorPhaseTransport? transport; + try + { + transport = await _transports.ResolveAsync(hostId, ct).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + throw; + } + catch (Exception ex) + { + throw new ExecutorPhaseTransportException(hostId, "resolve-transport", ex.Message, ex); + } + + if (transport is null) + throw new ExecutorPhaseTransportException(hostId, "resolve-transport", "No dispatch transport is configured for this host."); + return transport; + } + + private static async Task CallTransportAsync( + string hostId, + string operation, + Func> call, + CancellationToken ct) + { + try + { + return await call(ct).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + throw; + } + catch (ExecutorPhaseTransportException) + { + throw; + } + catch (ExecutorPhaseException) + { + // A stage-out transport that enforces the archive cap (or any + // phase-side rejection raised mid-transfer) reports a phase + // failure, not a host failure: the host was reachable. Let it + // through unwrapped so it is not charged as a transport failure. + throw; + } + catch (Exception ex) + { + throw new ExecutorPhaseTransportException(hostId, operation, ex.Message, ex); + } + } + + private static Task CallTransportAsync( + string hostId, + string operation, + Func call, + CancellationToken ct) => + CallTransportAsync(hostId, operation, async token => { await call(token).ConfigureAwait(false); return null; }, ct); + + private async Task SelectExecutorAsync(CancellationToken ct) + { + var workers = await _registry.ListAsync(ct).ConfigureAwait(false); + return workers + .Where(w => w.IsExecutor && w.ExecutorHostId is not null) + .Select(w => new ExecutorRegistration + { + HostId = w.ExecutorHostId!, + MaxConcurrentSandboxes = w.MaxConcurrentSandboxes, + AllowedNetworkProfiles = w.ExecutorNetworkProfiles ?? [], + DeclaredCredentials = w.ExecutorCredentials ?? [], + Cordoned = w.Cordoned, + Healthy = w.Healthy, + }) + .Where(reg => ExecutorEligibility.IsEligibleForPlacement(reg, currentLoad: 0)) + .OrderBy(reg => reg.HostId, StringComparer.Ordinal) + .Select(reg => reg.HostId) + .FirstOrDefault(); + } + + internal static string BuildDispatchKey(ExecutorPhaseRequest request) => + $"executor-phase/v1/{request.WorkItemId}/{request.Phase}/{request.Attempt}"; + + internal static string ComputeBodyHash(ExecutorPhaseRequest request) + { + static void WriteField(SHA256 sha, string value) + { + var bytes = Encoding.UTF8.GetBytes(value); + var lengthPrefix = Encoding.UTF8.GetBytes(bytes.Length.ToString(System.Globalization.CultureInfo.InvariantCulture) + ":"); + sha.TransformBlock(lengthPrefix, 0, lengthPrefix.Length, null, 0); + sha.TransformBlock(bytes, 0, bytes.Length, null, 0); + } + + using var sha = SHA256.Create(); + WriteField(sha, request.WorkItemId); + WriteField(sha, request.Phase); + WriteField(sha, request.Attempt.ToString(System.Globalization.CultureInfo.InvariantCulture)); + WriteField(sha, request.RepositoryId); + WriteField(sha, request.PayloadJson ?? string.Empty); + sha.TransformFinalBlock([], 0, 0); + return Convert.ToHexString(sha.Hash!).ToLowerInvariant(); + } + + internal static void ValidateRequest(ExecutorPhaseRequest request, ExecutorPhaseDispatchOptions options) + { + ArgumentNullException.ThrowIfNull(request); + ArgumentNullException.ThrowIfNull(options); + if (string.IsNullOrWhiteSpace(request.WorkItemId) || request.WorkItemId.Length > MaxWorkItemIdLength) + throw new ArgumentException($"WorkItemId must be 1..{MaxWorkItemIdLength} chars.", nameof(request)); + if (request.Phase is null || !PhaseNamePattern.IsMatch(request.Phase)) + throw new ArgumentException("Phase must match [A-Za-z0-9_-]{1,64}.", nameof(request)); + if (request.Attempt < 0) + throw new ArgumentOutOfRangeException(nameof(request), "Attempt must be zero or positive."); + if (string.IsNullOrWhiteSpace(request.RepositoryId) || request.RepositoryId.Length > MaxRepositoryIdLength) + throw new ArgumentException($"RepositoryId must be 1..{MaxRepositoryIdLength} chars.", nameof(request)); + var payloadBytes = Encoding.UTF8.GetByteCount(request.PayloadJson ?? string.Empty); + if (payloadBytes > options.MaxRequestPayloadBytes) + throw new ArgumentException($"PayloadJson exceeds MaxRequestPayloadBytes={options.MaxRequestPayloadBytes}.", nameof(request)); + } + + internal static ExecutorPhaseResult ValidateResult(ExecutorPhaseResult result, ExecutorPhaseDispatchOptions options) + { + ArgumentNullException.ThrowIfNull(result); + ArgumentNullException.ThrowIfNull(options); + if (!Enum.IsDefined(result.Outcome)) + throw new ExecutorPhaseException("Executor returned an unknown phase outcome."); + if (result.Findings is null) + throw new ExecutorPhaseException("Executor returned no findings collection."); + if (result.Findings.Count > options.MaxResultFindings) + throw new ExecutorPhaseException($"Executor returned {result.Findings.Count} findings, exceeding MaxResultFindings={options.MaxResultFindings}."); + foreach (var finding in result.Findings) + { + if (finding is null || finding.Length > options.MaxFindingLengthChars) + throw new ExecutorPhaseException($"Executor returned an oversized finding (limit {options.MaxFindingLengthChars} chars)."); + } + if (!string.IsNullOrEmpty(result.CommitSha) && !IsHexSha(result.CommitSha)) + throw new ExecutorPhaseException("Executor returned a malformed commit sha."); + if (result.Usage is null) + throw new ExecutorPhaseException("Executor returned no usage."); + if (result.Usage.InputTokens < 0 || result.Usage.OutputTokens < 0 || result.Usage.CostUsd < 0) + throw new ExecutorPhaseException("Executor returned negative usage."); + if (result.ErrorMessage is not null && result.ErrorMessage.Length > options.MaxResultErrorLengthChars) + throw new ExecutorPhaseException($"Executor returned an oversized error message (limit {options.MaxResultErrorLengthChars} chars)."); + return result; + } + + internal static string CanonicalizeRepoPath(string repoPath, string rootDirectory, string repositoryId) + { + if (string.IsNullOrWhiteSpace(repoPath)) + throw new ExecutorPhaseException($"Bare repo path for '{repositoryId}' is empty."); + string canonical; + try + { + canonical = Path.GetFullPath(repoPath); + } + catch (Exception ex) when (ex is ArgumentException or NotSupportedException or PathTooLongException) + { + throw new ExecutorPhaseException($"Bare repo path for '{repositoryId}' is not a valid path.", ex); + } + + if (!string.IsNullOrWhiteSpace(rootDirectory)) + { + string canonicalRoot; + try + { + canonicalRoot = Path.GetFullPath(rootDirectory); + } + catch (Exception ex) when (ex is ArgumentException or NotSupportedException or PathTooLongException) + { + throw new ExecutorPhaseException("Repositories root directory is not a valid path.", ex); + } + + var prefix = canonicalRoot.TrimEnd(Path.DirectorySeparatorChar) + Path.DirectorySeparatorChar; + if (!canonical.StartsWith(prefix, StringComparison.Ordinal)) + throw new ExecutorPhaseException($"Bare repo path for '{repositoryId}' escapes the repositories root."); + } + + return canonical; + } + + private static byte[] SerializeResult(ExecutorPhaseResult result) => + JsonSerializer.SerializeToUtf8Bytes(result, JsonOptions); + + private static ExecutorPhaseResult DeserializeCachedResult(string dispatchKey, IdempotencyEntry entry) + { + ExecutorPhaseResult? result; + try + { + result = JsonSerializer.Deserialize(entry.ResponseBody, JsonOptions); + } + catch (JsonException ex) + { + throw new ExecutorPhaseException($"Cached dispatch result for '{dispatchKey}' is corrupt.", ex); + } + + if (result is null) + throw new ExecutorPhaseException($"Cached dispatch result for '{dispatchKey}' is corrupt."); + return result; + } + + private static bool IsHexSha(string value) + { + if (value.Length is not (40 or 64)) + return false; + foreach (var ch in value) + { + if (!Uri.IsHexDigit(ch)) + return false; + } + return true; + } +} diff --git a/src/CodeyBox.Orchestrator/ExecutorPhaseRunner.cs b/src/CodeyBox.Orchestrator/ExecutorPhaseRunner.cs new file mode 100644 index 00000000..095f59cb --- /dev/null +++ b/src/CodeyBox.Orchestrator/ExecutorPhaseRunner.cs @@ -0,0 +1,87 @@ +using CodeyBox.Core; + +namespace CodeyBox.Orchestrator; + +/// +/// Executes one dispatched phase against the bare repo at +/// and returns its result. The in-process runner +/// calls this against the orchestrator's own repo path; a remote executor +/// runs the same logic against its staged copy, which is what makes a remote +/// dispatch return an outcome equivalent to running it in process. +/// Executor-side wiring of this seam (through the executor host process) is a +/// follow-up; the proxy and its tests run it directly through the transport +/// fake's staged copy. +/// +public interface IExecutorPhaseHandler +{ + Task ExecuteAsync( + ExecutorPhaseRequest request, + string repoPath, + CancellationToken ct); +} + +/// +/// Phase-execution interface for one work-item phase. Implemented by +/// (runs the phase against the +/// orchestrator's own bare repo) and +/// (dispatches to a registered executor when one is available, otherwise +/// falls back to the in-process runner with unchanged behaviour). +/// +public interface IExecutorPhaseRunner +{ + Task ExecutePhaseAsync(ExecutorPhaseRequest request, CancellationToken ct); +} + +/// +/// In-process : resolves the phase's bare +/// repo through and runs the injected +/// against it. Used directly when no +/// executor is registered and as the proxy's fallback. +/// +public sealed class InProcessExecutorPhaseRunner : IExecutorPhaseRunner +{ + private readonly IGitHost _gitHost; + private readonly IExecutorPhaseHandler _handler; + private readonly Func? _optionsAccessor; + + public InProcessExecutorPhaseRunner( + IGitHost gitHost, + IExecutorPhaseHandler handler, + Func? optionsAccessor = null) + { + _gitHost = gitHost ?? throw new ArgumentNullException(nameof(gitHost)); + _handler = handler ?? throw new ArgumentNullException(nameof(handler)); + _optionsAccessor = optionsAccessor; + } + + public async Task ExecutePhaseAsync(ExecutorPhaseRequest request, CancellationToken ct) + { + ExecutorPhaseProxy.ValidateRequest(request, ResolvedOptions()); + var repoPath = ResolveRepoPath(request.RepositoryId); + var result = await _handler.ExecuteAsync(request, repoPath, ct).ConfigureAwait(false); + return ExecutorPhaseProxy.ValidateResult(result, ResolvedOptions()); + } + + private string ResolveRepoPath(string repositoryId) + { + string repoPath; + try + { + repoPath = _gitHost.GetRepoPath(repositoryId); + } + catch (NotSupportedException ex) + { + throw new InvalidOperationException( + $"Git host exposes no local repository path for '{repositoryId}'; in-process phase execution needs a filesystem-backed repo.", ex); + } + + return ExecutorPhaseProxy.CanonicalizeRepoPath(repoPath, _gitHost.RepositoriesRootDirectory, repositoryId); + } + + private ExecutorPhaseDispatchOptions ResolvedOptions() + { + var options = _optionsAccessor?.Invoke() ?? new ExecutorPhaseDispatchOptions(); + options.Validate(); + return options; + } +} diff --git a/src/CodeyBox.Orchestrator/ExecutorStageOutValidator.cs b/src/CodeyBox.Orchestrator/ExecutorStageOutValidator.cs new file mode 100644 index 00000000..23e72efc --- /dev/null +++ b/src/CodeyBox.Orchestrator/ExecutorStageOutValidator.cs @@ -0,0 +1,337 @@ +using System.Formats.Tar; +using CodeyBox.Core; + +namespace CodeyBox.Orchestrator; + +/// +/// Proxy-side validation for the tar archive an executor stages back. +/// Mirrors the safety properties of the SSH transport's own stage-out +/// validation (bounded archive bytes, bounded entry count, bounded declared +/// payload vs archive size, path containment under the expected repo root, +/// directory/regular-file entries only): the remote payload is untrusted and +/// nothing is extracted over the orchestrator's bare repo until every check +/// passes. The archive-byte cap is enforced first by the transport while +/// receiving (so an unbounded payload cannot fill the orchestrator disk); +/// the size check here is defense in depth for transports that landed the +/// file by other means. Violations throw — the host +/// was reachable, so this is a phase failure, not a transport failure — and +/// leave the bare repo untouched. +/// +public static class ExecutorStageOutValidator +{ + private const int CopyBufferSize = 128 * 1024; + + /// + /// Validates the archive at , extracts it + /// into a fresh directory under , and + /// atomically swaps the validated repo tree into + /// . The expected single root entry is + /// the target's basename (for example item-id.git). + /// + public static async Task ValidateAndInstallAsync( + string archivePath, + string targetRepoPath, + string scratchRoot, + ExecutorPhaseDispatchOptions options, + CancellationToken ct) + { + ArgumentException.ThrowIfNullOrWhiteSpace(archivePath); + ArgumentException.ThrowIfNullOrWhiteSpace(targetRepoPath); + ArgumentException.ThrowIfNullOrWhiteSpace(scratchRoot); + ArgumentNullException.ThrowIfNull(options); + + string canonicalTarget; + string canonicalScratch; + try + { + canonicalTarget = Path.GetFullPath(targetRepoPath); + canonicalScratch = Path.GetFullPath(scratchRoot); + } + catch (Exception ex) when (ex is ArgumentException or NotSupportedException or PathTooLongException) + { + throw new ExecutorPhaseException($"Invalid stage-out path: {ex.Message}", ex); + } + + if (!Path.IsPathRooted(canonicalTarget) || !Path.IsPathRooted(canonicalScratch)) + throw new ExecutorPhaseException("Stage-out target and scratch paths must be absolute."); + + var tempRoot = Path.GetFullPath(Path.GetTempPath()); + var tempPrefix = tempRoot.TrimEnd(Path.DirectorySeparatorChar) + Path.DirectorySeparatorChar; + var scratchWithSep = canonicalScratch.TrimEnd(Path.DirectorySeparatorChar) + Path.DirectorySeparatorChar; + if (!scratchWithSep.StartsWith(tempPrefix, StringComparison.Ordinal) && !string.Equals(canonicalScratch.TrimEnd(Path.DirectorySeparatorChar), tempRoot.TrimEnd(Path.DirectorySeparatorChar), StringComparison.Ordinal)) + throw new ExecutorPhaseException("Stage-out scratch root escapes the temp directory."); + + var expectedRoot = Path.GetFileName(canonicalTarget.TrimEnd(Path.DirectorySeparatorChar)); + if (string.IsNullOrEmpty(expectedRoot)) + throw new ExecutorPhaseException($"Target repo path has no basename: '{targetRepoPath}'."); + + var archiveBytes = new FileInfo(archivePath).Length; + if (archiveBytes > options.StageOutMaxArchiveBytes) + throw new ExecutorPhaseException( + $"Staged-back archive for '{expectedRoot}' is {archiveBytes} bytes, exceeding the configured StageOutMaxArchiveBytes={options.StageOutMaxArchiveBytes}."); + + ValidateArchiveEntries(archivePath, archiveBytes, expectedRoot, options); + + var workRoot = Path.Combine(canonicalScratch, ".codeybox-stageout-" + Guid.NewGuid().ToString("N")); + var extractRoot = Path.Combine(workRoot, "extract"); + Directory.CreateDirectory(extractRoot); + try + { + await ExtractArchiveAsync(archivePath, extractRoot, expectedRoot, ct).ConfigureAwait(false); + var extracted = Path.Combine(extractRoot, expectedRoot); + if (File.Exists(extracted)) + throw new ExecutorPhaseException( + $"Staged-back archive root '{expectedRoot}' is a file, expected a directory."); + if (!Directory.Exists(extracted)) + throw new ExecutorPhaseException( + $"Validated staged-back archive did not contain expected root '{expectedRoot}'."); + ReplacePath(extracted, canonicalTarget); + } + finally + { + try { if (Directory.Exists(workRoot)) Directory.Delete(workRoot, recursive: true); } + catch { } + } + } + + private static void ValidateArchiveEntries( + string archivePath, + long archiveBytes, + string expectedRoot, + ExecutorPhaseDispatchOptions options) + { + var maxDeclaredBytes = MaxDeclaredPayloadBytes(archiveBytes, options.StageOutMaxExpansionRatio); + long declaredRegularFileBytes = 0; + var entryCount = 0; + var sawRootedEntry = false; + try + { + using var archive = File.OpenRead(archivePath); + using var reader = new TarReader(archive, leaveOpen: false); + TarEntry? entry; + while ((entry = reader.GetNextEntry(copyData: false)) is not null) + { + if (IsMetadataEntry(entry.EntryType)) + continue; + + entryCount++; + if (entryCount > options.StageOutMaxEntries) + throw new ExecutorPhaseException( + $"Staged-back archive exceeded configured StageOutMaxEntries={options.StageOutMaxEntries}."); + + if (!IsSafeEntryType(entry.EntryType)) + throw new ExecutorPhaseException( + $"Unsafe staged-back entry '{entry.Name}' has unsupported type '{entry.EntryType}'."); + + if (entry.EntryType is TarEntryType.RegularFile or TarEntryType.V7RegularFile) + { + if (entry.Length > maxDeclaredBytes - declaredRegularFileBytes) + throw new ExecutorPhaseException( + $"Staged-back archive declared file bytes exceeding StageOutMaxExpansionRatio={options.StageOutMaxExpansionRatio} for archive size {archiveBytes}."); + declaredRegularFileBytes += entry.Length; + } + + EnsureEntrySafe(entry, expectedRoot); + sawRootedEntry = true; + } + } + catch (ExecutorPhaseException) + { + throw; + } + catch (Exception ex) when (ex is InvalidDataException or IOException) + { + throw new ExecutorPhaseException($"Staged-back archive failed validation: {ex.Message}", ex); + } + + if (!sawRootedEntry) + throw new ExecutorPhaseException("Staged-back archive contained no extractable entries."); + } + + private static async Task ExtractArchiveAsync( + string archivePath, + string extractRoot, + string expectedRoot, + CancellationToken ct) + { + var canonicalRoot = Path.GetFullPath(extractRoot); + try + { + await using var archive = File.OpenRead(archivePath); + using var reader = new TarReader(archive, leaveOpen: false); + TarEntry? entry; + while ((entry = reader.GetNextEntry(copyData: false)) is not null) + { + if (IsMetadataEntry(entry.EntryType)) + continue; + var name = EnsureEntrySafe(entry, expectedRoot); + var destination = Path.GetFullPath(Path.Combine(canonicalRoot, name)); + EnsureContained(canonicalRoot, destination); + if (entry.EntryType == TarEntryType.Directory) + { + Directory.CreateDirectory(destination); + } + else + { + var parent = Path.GetDirectoryName(destination); + if (parent is not null) + Directory.CreateDirectory(parent); + await using var output = new FileStream( + destination, FileMode.Create, FileAccess.Write, FileShare.None, + bufferSize: CopyBufferSize, useAsync: true); + if (entry.DataStream is not null) + await BoundedCopyAsync(entry.DataStream, output, entry.Length, name, ct).ConfigureAwait(false); + } + } + } + catch (ExecutorPhaseException) + { + throw; + } + catch (Exception ex) when (ex is InvalidDataException or IOException) + { + throw new ExecutorPhaseException($"Staged-back archive failed extraction: {ex.Message}", ex); + } + } + + private static async Task BoundedCopyAsync(Stream source, Stream destination, long declaredLength, string name, CancellationToken ct) + { + var buffer = new byte[CopyBufferSize]; + long copied = 0; + while (true) + { + var read = await source.ReadAsync(buffer, ct).ConfigureAwait(false); + if (read == 0) + return; + copied += read; + if (copied > declaredLength) + throw new ExecutorPhaseException($"Staged-back entry '{name}' streamed more bytes than its declared length."); + await destination.WriteAsync(buffer.AsMemory(0, read), ct).ConfigureAwait(false); + } + } + + private static long MaxDeclaredPayloadBytes(long archiveBytes, double maxExpansionRatio) + { + var capped = archiveBytes * maxExpansionRatio; + if (double.IsInfinity(capped) || capped >= long.MaxValue) + return long.MaxValue; + return Math.Max(archiveBytes, (long)Math.Ceiling(capped)); + } + + private static bool IsSafeEntryType(TarEntryType type) => + type is TarEntryType.Directory + or TarEntryType.RegularFile + or TarEntryType.V7RegularFile; + + private static bool IsMetadataEntry(TarEntryType type) => + type is TarEntryType.ExtendedAttributes + or TarEntryType.GlobalExtendedAttributes; + + internal static string NormalizeEntryName(string name) + { + if (string.IsNullOrWhiteSpace(name)) + throw new ExecutorPhaseException("Staged-back archive contains an entry with an empty name."); + if (name.IndexOf('\0') >= 0) + throw new ExecutorPhaseException("Staged-back archive contains an entry with a NUL byte in its name."); + + var normalized = name.Replace('\\', '/'); + while (normalized.StartsWith("./", StringComparison.Ordinal)) + normalized = normalized[2..]; + normalized = normalized.TrimEnd('/'); + if (normalized.Length == 0 || normalized[0] == '/') + throw new ExecutorPhaseException($"Unsafe staged-back entry path '{name}'."); + + foreach (var part in normalized.Split('/')) + { + if (part.Length == 0 || part == "." || part == "..") + throw new ExecutorPhaseException($"Unsafe staged-back entry path '{name}'."); + } + + return normalized; + } + + private static string EnsureEntrySafe(TarEntry entry, string expectedRoot) + { + if (!IsSafeEntryType(entry.EntryType)) + throw new ExecutorPhaseException( + $"Unsafe staged-back entry '{entry.Name}' has unsupported type '{entry.EntryType}'."); + var name = NormalizeEntryName(entry.Name); + EnsureUnderExpectedRoot(name, expectedRoot); + if (string.Equals(name, expectedRoot, StringComparison.Ordinal) + && entry.EntryType != TarEntryType.Directory) + throw new ExecutorPhaseException( + $"Staged-back archive root '{expectedRoot}' must be a directory, not '{entry.EntryType}'."); + return name; + } + + private static void EnsureUnderExpectedRoot(string entryName, string expectedRoot) + { + if (string.Equals(entryName, expectedRoot, StringComparison.Ordinal)) + return; + if (entryName.StartsWith(expectedRoot + "/", StringComparison.Ordinal)) + return; + throw new ExecutorPhaseException( + $"Unsafe staged-back entry '{entryName}' is outside expected root '{expectedRoot}'."); + } + + private static void EnsureContained(string root, string candidate) + { + var normalizedRoot = root.TrimEnd(Path.DirectorySeparatorChar) + Path.DirectorySeparatorChar; + if (!candidate.StartsWith(normalizedRoot, StringComparison.Ordinal)) + throw new ExecutorPhaseException( + $"Unsafe staged-back entry escapes staging root."); + } + + private static void ReplacePath(string source, string target) + { + var backup = target + ".codeybox-backup-" + Guid.NewGuid().ToString("N"); + var hadTarget = Path.Exists(target); + if (hadTarget) + MovePath(target, backup); + + try + { + MovePath(source, target); + } + catch + { + if (hadTarget) + { + try + { + if (!Path.Exists(target) && Path.Exists(backup)) + MovePath(backup, target); + } + catch { } + } + throw; + } + + if (hadTarget) + { + try + { + if (Directory.Exists(backup)) + Directory.Delete(backup, recursive: true); + else if (File.Exists(backup)) + File.Delete(backup); + } + catch { } + } + } + + private static void MovePath(string source, string target) + { + try + { + if (Directory.Exists(source)) + Directory.Move(source, target); + else + File.Move(source, target); + } + catch (IOException ex) + { + throw new ExecutorPhaseException($"Failed to install validated staged-back payload: {ex.Message}", ex); + } + } +} diff --git a/tests/CodeyBox.Tests/ExecutorPhaseProxyTests.cs b/tests/CodeyBox.Tests/ExecutorPhaseProxyTests.cs new file mode 100644 index 00000000..e566d21a --- /dev/null +++ b/tests/CodeyBox.Tests/ExecutorPhaseProxyTests.cs @@ -0,0 +1,829 @@ +using System.Diagnostics; +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 phase-dispatch proxy: remote execution +/// equivalence, per-item repo staging and stage-back, stage-out bounds, +/// idempotent redelivery, transport-vs-agent failure classification, and +/// in-process fallback. Uses a real , a real +/// , real tar archives and real git +/// commits; only the network hop to the executor is faked (a tar-based +/// loopback transport over a per-host directory). +/// +public sealed class ExecutorPhaseProxyTests : IDisposable +{ + private const string FixedDate = "2000-01-01T00:00:00+00:00"; + + private readonly string _root = Directory.CreateTempSubdirectory("codeybox-exec-phase-").FullName; + + public void Dispose() + { + try { Directory.Delete(_root, recursive: true); } catch { } + } + + // ── verification 1: remote dispatch ≡ in-process ──────────────────────── + + [Fact] + public async Task RemoteDispatch_ReturnsOutcomeShaFindingsAndUsage_EquivalentToInProcess() + { + using var ctx = CreateContext(["exec-1"]); + var item = WorkItemId.New(); + var twin = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + await SeedBareRepoAsync(ctx.Git, twin); + + var remote = await ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0), CancellationToken.None); + var inner = await ctx.Inner.ExecutePhaseAsync(NewRequest(twin, "work", 0), CancellationToken.None); + + Assert.Equal(ExecutorPhaseOutcome.Succeeded, remote.Outcome); + AssertResultsEqual(inner, remote); + } + + // ── verification 2: commits land in the bare repo; push path unchanged ── + + [Fact] + public async Task RemoteCommit_LandsInBareRepo_AndPushPublishesItUnchanged() + { + using var ctx = CreateContext(["exec-1"]); + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + + var result = await ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0), CancellationToken.None); + Assert.NotNull(result.CommitSha); + + var bare = ctx.Git.GetRepoPath(item.ToString()); + var log = await RunGitBareCapture(bare, "log", "--format=%H", "phase/work-0"); + Assert.Contains(result.CommitSha!, log.Split('\n', StringSplitOptions.RemoveEmptyEntries)); + + var upstream = Path.Combine(_root, "upstream-" + Guid.NewGuid().ToString("N") + ".git"); + await RunGit(_root, "init", "--bare", upstream); + await RunGitBare(bare, "push", upstream, "phase/work-0:refs/heads/published"); + var published = await RunGitBareCapture(upstream, "rev-parse", "refs/heads/published"); + Assert.Equal(result.CommitSha, published.Trim()); + } + + // ── verification 3: per-item scoping ──────────────────────────────────── + + [Fact] + public async Task Executor_ReceivesOnlyTheStagedRepoForItsItem() + { + using var ctx = CreateContext(["exec-1"]); + var itemA = WorkItemId.New(); + var itemB = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, itemA); + await SeedBareRepoAsync(ctx.Git, itemB); + + await ctx.Proxy.ExecutePhaseAsync(NewRequest(itemA, "work", 0), CancellationToken.None); + + var transport = ctx.Transports["exec-1"]; + Assert.Single(transport.StagedInPaths); + Assert.Equal(Path.GetFullPath(ctx.Git.GetRepoPath(itemA.ToString())), transport.StagedInPaths[0]); + Assert.DoesNotContain(Path.GetFullPath(ctx.Git.GetRepoPath(itemB.ToString())), transport.StagedInPaths); + Assert.Single(Directory.GetFileSystemEntries(transport.ExecutorRoot)); + } + + // ── verification 4: stage-out bounds ──────────────────────────────────── + + [Fact] + public async Task StageBack_LargerThanMaxArchiveBytes_IsRejectedWithoutRepoWrite() + { + using var ctx = CreateContext(["exec-1"], maxArchiveBytes: 256 * 1024); + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + var bare = ctx.Git.GetRepoPath(item.ToString()); + var before = (await RunGitBareCapture(bare, "rev-parse", "phase/seed")).Trim(); + + ctx.Handler.PlantUnpackedBytes = 1024 * 1024; + var request = NewRequest(item, "work", 0); + await Assert.ThrowsAsync( + () => ctx.Proxy.ExecutePhaseAsync(request, CancellationToken.None)); + + // The cap travels into the transport and aborts the transfer + // mid-stream: fewer than the planted 1 MiB ever land on disk. + Assert.Equal(256 * 1024, ctx.Transports["exec-1"].LastStageOutMaxBytes); + Assert.True( + ctx.Transports["exec-1"].StageOutBytesWritten <= 256 * 1024, + $"Streaming stage-out wrote {ctx.Transports["exec-1"].StageOutBytesWritten} bytes past the 256 KiB cap."); + Assert.Equal(before, (await RunGitBareCapture(bare, "rev-parse", "phase/seed")).Trim()); + Assert.Empty((await RunGitBareCapture(bare, "branch", "--list", "phase/work-0")).Trim()); + var lookup = await ctx.Store.LookupAsync( + ExecutorPhaseProxy.BuildDispatchKey(request), + ExecutorPhaseProxy.ComputeBodyHash(request), + DateTimeOffset.UtcNow); + Assert.Equal(IdempotencyLookupOutcome.Miss, lookup.Outcome); + } + + [Fact] + public async Task StageBack_MoreEntriesThanAllowed_IsRejectedWithoutRepoWrite() + { + using var ctx = CreateContext(["exec-1"], maxEntries: 16); + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + var bare = ctx.Git.GetRepoPath(item.ToString()); + var before = (await RunGitBareCapture(bare, "rev-parse", "phase/seed")).Trim(); + + ctx.Handler.PlantFileCount = 40; + await Assert.ThrowsAsync( + () => ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0), CancellationToken.None)); + + Assert.Equal(before, (await RunGitBareCapture(bare, "rev-parse", "phase/seed")).Trim()); + Assert.Empty((await RunGitBareCapture(bare, "branch", "--list", "phase/work-0")).Trim()); + } + + [Fact] + public async Task StageBack_InflatedDeclaredSize_IsRejectedWithoutRepoWrite() + { + using var ctx = CreateContext(["exec-1"]); + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + var bare = ctx.Git.GetRepoPath(item.ToString()); + var before = (await RunGitBareCapture(bare, "rev-parse", "phase/seed")).Trim(); + + var rootName = Path.GetFileName(bare.TrimEnd(Path.DirectorySeparatorChar)); + ctx.Transports["exec-1"].CustomArchive = stream => + { + WriteInflatedArchive(stream, rootName); + return Task.CompletedTask; + }; + await Assert.ThrowsAsync( + () => ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0), CancellationToken.None)); + + Assert.Equal(before, (await RunGitBareCapture(bare, "rev-parse", "phase/seed")).Trim()); + Assert.Empty((await RunGitBareCapture(bare, "branch", "--list", "phase/work-0")).Trim()); + } + + [Theory] + [InlineData("traversal")] + [InlineData("absolute")] + [InlineData("wrong-root")] + [InlineData("symlink")] + public async Task StageBack_MaliciousEntry_IsRejectedWithoutRepoWrite(string kind) + { + using var ctx = CreateContext(["exec-1"]); + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + var bare = ctx.Git.GetRepoPath(item.ToString()); + var before = (await RunGitBareCapture(bare, "rev-parse", "phase/seed")).Trim(); + var rootName = Path.GetFileName(bare.TrimEnd(Path.DirectorySeparatorChar)); + + var (entryName, typeFlag) = kind switch + { + "traversal" => ("../evil.txt", '0'), + "absolute" => ("/tmp/evil.txt", '0'), + "wrong-root" => ("other-root.git/evil.txt", '0'), + "symlink" => (rootName + "/evil-link", '2'), + _ => throw new ArgumentOutOfRangeException(nameof(kind)), + }; + ctx.Transports["exec-1"].CustomArchive = stream => + { + WriteTarHeader(stream, entryName, typeFlag, 0); + return Task.CompletedTask; + }; + var request = NewRequest(item, "work", 0); + await Assert.ThrowsAsync( + () => ctx.Proxy.ExecutePhaseAsync(request, CancellationToken.None)); + + Assert.Equal(before, (await RunGitBareCapture(bare, "rev-parse", "phase/seed")).Trim()); + Assert.Empty((await RunGitBareCapture(bare, "branch", "--list", "phase/work-0")).Trim()); + Assert.False(File.Exists(Path.Combine(Path.GetDirectoryName(bare)!, "evil.txt")), "Traversal entry escaped the staging root."); + var lookup = await ctx.Store.LookupAsync( + ExecutorPhaseProxy.BuildDispatchKey(request), + ExecutorPhaseProxy.ComputeBodyHash(request), + DateTimeOffset.UtcNow); + Assert.Equal(IdempotencyLookupOutcome.Miss, lookup.Outcome); + } + + // ── verification 5: idempotency ───────────────────────────────────────── + + [Fact] + public async Task Redelivery_ReturnsOriginalResult_WithoutSecondSandbox() + { + using var ctx = CreateContext(["exec-1"]); + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + var request = NewRequest(item, "work", 0); + + var first = await ctx.Proxy.ExecutePhaseAsync(request, CancellationToken.None); + var second = await ctx.Proxy.ExecutePhaseAsync(request, CancellationToken.None); + + AssertResultsEqual(first, second); + Assert.Equal(1, ctx.Transports["exec-1"].RunPhaseCalls); + } + + [Fact] + public async Task SameKey_DifferentBody_IsRefusedAsConflictWithoutExecuting() + { + using var ctx = CreateContext(["exec-1"]); + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + + await ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0, "{\"v\":1}"), CancellationToken.None); + await Assert.ThrowsAsync( + () => ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0, "{\"v\":2}"), CancellationToken.None)); + + Assert.Equal(1, ctx.Transports["exec-1"].RunPhaseCalls); + } + + // ── verification 6: transport vs agent failure ────────────────────────── + + [Fact] + public async Task TransportFailure_PropagatesWithoutCachingOrRepoWrite() + { + using var ctx = CreateContext(["exec-1"]); + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + var bare = ctx.Git.GetRepoPath(item.ToString()); + var before = (await RunGitBareCapture(bare, "rev-parse", "phase/seed")).Trim(); + + ctx.Transports["exec-1"].FailWith = new ExecutorPhaseTransportException("exec-1", "stage-in", "connection refused"); + var request = NewRequest(item, "work", 0); + var thrown = await Assert.ThrowsAsync( + () => ctx.Proxy.ExecutePhaseAsync(request, CancellationToken.None)); + Assert.Equal("stage-in", thrown.Operation); + + Assert.Equal(before, (await RunGitBareCapture(bare, "rev-parse", "phase/seed")).Trim()); + Assert.Equal(0, ctx.InnerSpy.Calls); + var lookup = await ctx.Store.LookupAsync( + ExecutorPhaseProxy.BuildDispatchKey(request), + ExecutorPhaseProxy.ComputeBodyHash(request), + DateTimeOffset.UtcNow); + Assert.Equal(IdempotencyLookupOutcome.Miss, lookup.Outcome); + } + + [Fact] + public async Task AgentFailure_IsReturnedAndCached_NotThrown() + { + using var ctx = CreateContext(["exec-1"]); + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + + ctx.Handler.ForceAgentFailure = true; + var request = NewRequest(item, "work", 0, "{\"v\":9}"); + var first = await ctx.Proxy.ExecutePhaseAsync(request, CancellationToken.None); + Assert.Equal(ExecutorPhaseOutcome.AgentFailed, first.Outcome); + + var second = await ctx.Proxy.ExecutePhaseAsync(request, CancellationToken.None); + AssertResultsEqual(first, second); + Assert.Equal(1, ctx.Transports["exec-1"].RunPhaseCalls); + } + + // ── verification 7: fallback ──────────────────────────────────────────── + + [Fact] + public async Task NoExecutorRegistered_FallsBackToInProcess_Unchanged() + { + using var ctx = CreateContext([]); + var item = WorkItemId.New(); + var twin = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + await SeedBareRepoAsync(ctx.Git, twin); + + var fallback = await ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0), CancellationToken.None); + var direct = await ctx.Inner.ExecutePhaseAsync(NewRequest(twin, "work", 0), CancellationToken.None); + + AssertResultsEqual(direct, fallback); + Assert.Equal(0, ctx.Factory.Resolves); + } + + [Fact] + public async Task CordonedExecutor_IsNeverSelected_FallsBackToInProcess() + { + using var ctx = CreateContext(["exec-1"], cordoned: true); + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + + var result = await ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0), CancellationToken.None); + + Assert.Equal(ExecutorPhaseOutcome.Succeeded, result.Outcome); + Assert.NotNull(result.CommitSha); + Assert.Equal(0, ctx.Factory.Resolves); + Assert.Equal(1, ctx.InnerSpy.Calls); + } + + [Fact] + public void DispatchOptions_Validate_RejectsBadBounds() + { + Assert.Throws(() => new ExecutorPhaseDispatchOptions { StageOutMaxArchiveBytes = 0 }.Validate()); + Assert.Throws(() => new ExecutorPhaseDispatchOptions { StageOutMaxEntries = 0 }.Validate()); + Assert.Throws(() => new ExecutorPhaseDispatchOptions { StageOutMaxExpansionRatio = 0.5 }.Validate()); + new ExecutorPhaseDispatchOptions().Validate(); + } + + // ── harness ───────────────────────────────────────────────────────────── + + private static ExecutorPhaseRequest NewRequest(WorkItemId item, string phase, int attempt, string payload = "{}") => + new() { WorkItemId = item.ToString(), Phase = phase, Attempt = attempt, RepositoryId = item.ToString(), PayloadJson = payload }; + + private static void AssertResultsEqual(ExecutorPhaseResult expected, ExecutorPhaseResult actual) + { + Assert.Equal(expected.Outcome, actual.Outcome); + Assert.Equal(expected.CommitSha, actual.CommitSha); + Assert.Equal(expected.Findings, actual.Findings); + Assert.Equal(expected.Usage, actual.Usage); + Assert.Equal(expected.ErrorMessage, actual.ErrorMessage); + } + + private TestHarness CreateContext(string[] executors, long? maxArchiveBytes = null, int? maxEntries = null, bool cordoned = false) + { + 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 store = new SqliteIdempotencyStore(dbPath); + var registry = new FakeWorkerRegistry(); + var handler = new GitCommitPhaseHandler(); + var options = new ExecutorPhaseDispatchOptions(); + if (maxArchiveBytes is not null) options.StageOutMaxArchiveBytes = maxArchiveBytes.Value; + if (maxEntries is not null) options.StageOutMaxEntries = maxEntries.Value; + var factory = new FakeTransportFactory(); + var inner = new InProcessExecutorPhaseRunner(git, handler, () => options); + var spy = new SpyRunner(inner); + var proxy = new ExecutorPhaseProxy(registry, factory, git, store, spy, () => options); + foreach (var host in executors) + { + registry.AddExecutor(host, cordoned); + factory.AddHost(host, Path.Combine(_root, "executor-" + host + "-" + Guid.NewGuid().ToString("N")), handler); + } + return new TestHarness(git, store, handler, factory, inner, spy, proxy); + } + + private async Task SeedBareRepoAsync(LocalGitHost git, WorkItemId item) + { + var repoId = await git.EnsureRepositoryAsync(item, seedFromUrl: null); + var bare = git.GetRepoPath(repoId); + var clone = Path.Combine(_root, "seedclone-" + Guid.NewGuid().ToString("N")); + await RunGit(_root, "clone", bare, clone); + await RunGit(clone, "config", "user.email", "t@t"); + await RunGit(clone, "config", "user.name", "T"); + await File.WriteAllTextAsync(Path.Combine(clone, "README.md"), "seed\n"); + await RunGit(clone, "add", "README.md"); + await RunGit(clone, "commit", "-m", "seed"); + await RunGit(clone, "branch", "-M", "phase/seed"); + await RunGit(clone, "push", "origin", "phase/seed"); + } + + private async Task RunGitBare(string bareRepo, params string[] args) => + await RunGit(_root, ["--git-dir", bareRepo, .. args]); + + private async Task RunGitBareCapture(string bareRepo, params string[] args) => + await RunGitCapture(_root, ["--git-dir", bareRepo, .. args]); + + private static async Task RunGit(string cwd, params string[] args) + { + using var p = Process.Start(GitPsi(cwd, args))!; + var stderr = await p.StandardError.ReadToEndAsync(); + await p.StandardOutput.ReadToEndAsync(); + await p.WaitForExitAsync(); + if (p.ExitCode != 0) + throw new InvalidOperationException($"git {string.Join(' ', args)} failed: {stderr}"); + } + + private static async Task RunGitCapture(string cwd, params string[] args) + { + using var p = Process.Start(GitPsi(cwd, args))!; + var stdout = await p.StandardOutput.ReadToEndAsync(); + var stderr = await p.StandardError.ReadToEndAsync(); + await p.WaitForExitAsync(); + if (p.ExitCode != 0) + throw new InvalidOperationException($"git {string.Join(' ', args)} failed: {stderr}"); + return stdout; + } + + private static ProcessStartInfo GitPsi(string cwd, string[] args) + { + var psi = new ProcessStartInfo + { + FileName = "git", + WorkingDirectory = cwd, + RedirectStandardOutput = true, + RedirectStandardError = true, + UseShellExecute = false, + }; + foreach (var a in args) psi.ArgumentList.Add(a); + psi.Environment["GIT_AUTHOR_NAME"] = "CodeyBox Test"; + psi.Environment["GIT_AUTHOR_EMAIL"] = "test@codeybox.invalid"; + psi.Environment["GIT_COMMITTER_NAME"] = "CodeyBox Test"; + psi.Environment["GIT_COMMITTER_EMAIL"] = "test@codeybox.invalid"; + psi.Environment["GIT_AUTHOR_DATE"] = FixedDate; + psi.Environment["GIT_COMMITTER_DATE"] = FixedDate; + return psi; + } + + /// + /// Writes a tar whose file entry declares a 1 MiB payload while the + /// archive itself carries almost no data: declared bytes dwarf + /// archive-bytes × ratio, so the expansion-ratio guard must reject it. + /// + private static void WriteInflatedArchive(Stream stream, string rootName) + { + WriteTarHeader(stream, rootName + "/", '5', 0); + WriteTarHeader(stream, rootName + "/payload.bin", '0', 1024 * 1024); + stream.Write(new byte[1024]); + stream.Write(new byte[1024]); + } + + private static void WriteTarHeader(Stream stream, string name, char typeFlag, long size) + { + var block = new byte[512]; + var nameBytes = Encoding.ASCII.GetBytes(name); + Array.Copy(nameBytes, block, Math.Min(nameBytes.Length, 100)); + Encoding.ASCII.GetBytes("0000777\0").CopyTo(block, 100); + Encoding.ASCII.GetBytes("0000000\0").CopyTo(block, 108); + Encoding.ASCII.GetBytes("0000000\0").CopyTo(block, 116); + Encoding.ASCII.GetBytes(Convert.ToString(size, 8).PadLeft(11, '0') + "\0").CopyTo(block, 124); + Encoding.ASCII.GetBytes(Convert.ToString(946684800L, 8).PadLeft(11, '0') + "\0").CopyTo(block, 136); + for (var i = 148; i < 156; i++) block[i] = (byte)' '; + block[156] = (byte)typeFlag; + Encoding.ASCII.GetBytes("ustar\0" + "00").CopyTo(block, 257); + long checksum = 0; + foreach (var b in block) checksum += b; + Encoding.ASCII.GetBytes(Convert.ToString(checksum, 8).PadLeft(6, '0') + "\0 ").CopyTo(block, 148); + stream.Write(block); + } + + private sealed class TestHarness : IDisposable + { + public TestHarness( + LocalGitHost git, + SqliteIdempotencyStore store, + GitCommitPhaseHandler handler, + FakeTransportFactory factory, + InProcessExecutorPhaseRunner inner, + SpyRunner spy, + ExecutorPhaseProxy proxy) + { + Git = git; + Store = store; + Handler = handler; + Factory = factory; + Inner = inner; + InnerSpy = spy; + Proxy = proxy; + } + + public LocalGitHost Git { get; } + public SqliteIdempotencyStore Store { get; } + public GitCommitPhaseHandler Handler { get; } + public FakeTransportFactory Factory { get; } + public Dictionary Transports => Factory.Transports; + public InProcessExecutorPhaseRunner Inner { get; } + public SpyRunner InnerSpy { get; } + public ExecutorPhaseProxy Proxy { get; } + + public void Dispose() => Store.Dispose(); + } + + private sealed class SpyRunner : IExecutorPhaseRunner + { + private readonly IExecutorPhaseRunner _inner; + + public SpyRunner(IExecutorPhaseRunner inner) => _inner = inner; + + public int Calls { get; private set; } + + public Task ExecutePhaseAsync(ExecutorPhaseRequest request, CancellationToken ct) + { + Calls++; + return _inner.ExecutePhaseAsync(request, ct); + } + } + + private sealed class FakeWorkerRegistry : IWorkerRegistry + { + private readonly Dictionary _rows = new(StringComparer.Ordinal); + + public void AddExecutor(string hostId, bool cordoned = false) + { + 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 = cordoned, + 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); + } + + /// + /// Deterministic phase handler over real git: clones the given bare repo, + /// commits one file on phase/<phase>-<attempt> with fixed + /// identity and timestamps (so identical starting repos yield identical + /// shas), pushes back, and returns the sha with fixed findings and usage. + /// + private sealed class GitCommitPhaseHandler : IExecutorPhaseHandler + { + public int PlantUnpackedBytes; + public int PlantFileCount; + public bool ForceAgentFailure; + + public async Task ExecuteAsync(ExecutorPhaseRequest request, string repoPath, CancellationToken ct) + { + if (ForceAgentFailure) + { + return new ExecutorPhaseResult + { + Outcome = ExecutorPhaseOutcome.AgentFailed, + Usage = new ExecutorPhaseUsage(10, 5, 0.001m), + Findings = ["agent could not complete the phase"], + ErrorMessage = "simulated agent failure", + }; + } + + var work = Path.Combine(Path.GetTempPath(), "codeybox-phasework-" + Guid.NewGuid().ToString("N")); + Directory.CreateDirectory(work); + try + { + var branch = $"phase/{request.Phase}-{request.Attempt}"; + await RunGit(work, "clone", repoPath, "w"); + var w = Path.Combine(work, "w"); + await RunGit(w, "config", "user.email", "t@t"); + await RunGit(w, "config", "user.name", "T"); + await RunGit(w, "checkout", "-B", branch, "origin/phase/seed"); + // Content deliberately excludes the work item id so identical + // starting repos yield identical commits (and shas) whether + // the phase runs remotely or in process. + var content = $"phase={request.Phase} attempt={request.Attempt}\n"; + await File.WriteAllTextAsync(Path.Combine(w, "phase-output.txt"), content, ct); + for (var i = 0; i < PlantFileCount; i++) + await File.WriteAllTextAsync(Path.Combine(w, $"planted-{i}.txt"), "x\n", ct); + await RunGit(w, "add", "-A"); + await RunGit(w, "commit", "-m", $"phase {request.Phase} attempt {request.Attempt}"); + var sha = (await RunGitCapture(w, "rev-parse", "HEAD")).Trim(); + await RunGit(w, "push", "origin", $"{branch}:{branch}"); + if (PlantUnpackedBytes > 0) + { + // Incompressible payload written straight into the staged + // copy (outside git, as hostile executor content would + // be): git packs would otherwise compress planted zeros + // and the archive would stay under the cap. + var random = new Random(42); + var bytes = new byte[PlantUnpackedBytes]; + random.NextBytes(bytes); + await File.WriteAllBytesAsync(Path.Combine(repoPath, "planted-unpacked.bin"), bytes, ct); + } + return new ExecutorPhaseResult + { + Outcome = ExecutorPhaseOutcome.Succeeded, + CommitSha = sha, + Findings = [$"finding for {request.Phase}"], + Usage = new ExecutorPhaseUsage(100, 50, 0.002m), + }; + } + finally + { + try { Directory.Delete(work, recursive: true); } catch { } + } + } + } + + /// + /// Loopback executor transport: each host owns a directory standing in + /// for its machine. Stage-in copies the single repo path there; the phase + /// runs the shared handler against that copy; stage-out tars the copy + /// back through real tar bytes so the proxy validates a real archive. + /// + private sealed class FakePhaseTransport : IExecutorPhaseTransport + { + private readonly GitCommitPhaseHandler _handler; + + public FakePhaseTransport(string hostId, string executorRoot, GitCommitPhaseHandler handler) + { + HostId = hostId; + ExecutorRoot = executorRoot; + _handler = handler; + Directory.CreateDirectory(executorRoot); + } + + public string HostId { get; } + public string ExecutorRoot { get; } + public List StagedInPaths { get; } = []; + public int RunPhaseCalls { get; private set; } + public ExecutorPhaseTransportException? FailWith; + public Func? CustomArchive; + public long? LastStageOutMaxBytes { get; private set; } + public long StageOutBytesWritten { get; private set; } + + public string? StagedCopy + { + get + { + var entries = Directory.GetFileSystemEntries(ExecutorRoot); + return entries.Length == 1 ? entries[0] : null; + } + } + + public Task StageInAsync(string hostRepoPath, CancellationToken ct) + { + ThrowIfFailing(); + StagedInPaths.Add(Path.GetFullPath(hostRepoPath)); + var dest = Path.Combine(ExecutorRoot, Path.GetFileName(hostRepoPath.TrimEnd(Path.DirectorySeparatorChar))); + CopyDirectory(hostRepoPath, dest); + return Task.CompletedTask; + } + + public Task RunPhaseAsync(ExecutorPhaseRequest request, CancellationToken ct) + { + ThrowIfFailing(); + RunPhaseCalls++; + var staged = StagedCopy ?? throw new InvalidOperationException("No staged repo on fake executor."); + return _handler.ExecuteAsync(request, staged, ct); + } + + public async Task StageOutToArchiveAsync(string hostArchivePath, long maxArchiveBytes, CancellationToken ct) + { + ThrowIfFailing(); + LastStageOutMaxBytes = maxArchiveBytes; + StageOutBytesWritten = 0; + await using var file = File.OpenWrite(hostArchivePath); + await using var bounded = new BoundedStageOutStream(file, maxArchiveBytes, HostId, bytes => StageOutBytesWritten = bytes); + try + { + if (CustomArchive is not null) + { + await CustomArchive(bounded).ConfigureAwait(false); + await bounded.FlushAsync(ct).ConfigureAwait(false); + return; + } + var staged = StagedCopy ?? throw new InvalidOperationException("No staged repo on fake executor."); + await WriteTarOfDirectoryAsync(staged, bounded, ct).ConfigureAwait(false); + } + catch + { + // The streaming cap was exceeded (or the payload was hostile): + // leave no usable archive behind, mirroring the production + // contract that an aborted stage-out is a phase failure. + try { File.Delete(hostArchivePath); } catch { } + throw; + } + } + + private void ThrowIfFailing() + { + if (FailWith is not null) throw FailWith; + } + + 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); + } + } + + /// + /// Write-only wrapper that aborts the stage-out as soon as the + /// configured archive cap is exceeded, so the fake mirrors the + /// production contract: the cap is enforced while receiving, never + /// by measuring a fully-buffered file afterwards. Exceeding the cap + /// is an (phase failure: the + /// host was reachable) rather than a transport failure. + /// + private sealed class BoundedStageOutStream : Stream + { + private readonly Stream _inner; + private readonly long _maxBytes; + private readonly string _hostId; + private readonly Action _progress; + private long _written; + + public BoundedStageOutStream(Stream inner, long maxBytes, string hostId, Action progress) + { + _inner = inner; + _maxBytes = maxBytes; + _hostId = hostId; + _progress = progress; + } + + public override bool CanRead => false; + public override bool CanSeek => false; + public override bool CanWrite => true; + public override long Length => _written; + public override long Position { get => _written; set => throw new NotSupportedException(); } + + public override void Write(byte[] buffer, int offset, int count) + { + if (_written + count > _maxBytes) + throw new ExecutorPhaseException( + $"Staged-back archive from host '{_hostId}' exceeded configured StageOutMaxArchiveBytes={_maxBytes}."); + _inner.Write(buffer, offset, count); + _written += count; + _progress(_written); + } + + public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken ct) + { + if (_written + count > _maxBytes) + throw new ExecutorPhaseException( + $"Staged-back archive from host '{_hostId}' exceeded configured StageOutMaxArchiveBytes={_maxBytes}."); + await _inner.WriteAsync(buffer.AsMemory(offset, count), ct).ConfigureAwait(false); + _written += count; + _progress(_written); + } + + public override async ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken ct = default) + { + if (_written + buffer.Length > _maxBytes) + throw new ExecutorPhaseException( + $"Staged-back archive from host '{_hostId}' exceeded configured StageOutMaxArchiveBytes={_maxBytes}."); + await _inner.WriteAsync(buffer, ct).ConfigureAwait(false); + _written += buffer.Length; + _progress(_written); + } + + public override void Flush() => _inner.Flush(); + public override Task FlushAsync(CancellationToken ct) => _inner.FlushAsync(ct); + public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + public override void SetLength(long value) => throw new NotSupportedException(); + } + } + + private sealed class FakeTransportFactory : IExecutorPhaseTransportFactory + { + public readonly Dictionary Transports = new(StringComparer.Ordinal); + public int Resolves { get; private set; } + + public void AddHost(string hostId, string executorRoot, GitCommitPhaseHandler handler) => + Transports[hostId] = new FakePhaseTransport(hostId, executorRoot, handler); + + public Task ResolveAsync(string hostId, CancellationToken ct) + { + Resolves++; + return Task.FromResult(Transports.GetValueOrDefault(hostId)); + } + } +}