Dispatch a phase to a registered executor host, with the per-item bare repo staged in and out and delivery made idempotent

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
 

src/CodeyBox.Core/ExecutorPhase.cs

@@ -0,0 +1,88 @@
+using System.Text.Json.Serialization;
+
+namespace CodeyBox.Core;
+
+/// <summary>
+/// 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
+/// <see cref="ExecutorPhaseTransportException"/> so the caller retries
+/// elsewhere instead of charging the work item with an agent failure.
+/// </summary>
+public enum ExecutorPhaseOutcome
+{
+    Succeeded = 0,
+    AgentFailed = 1,
+}
+
+/// <summary>
+/// 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.
+/// </summary>
+public sealed record ExecutorPhaseUsage(
+    [property: JsonPropertyName("inputTokens")] long InputTokens,
+    [property: JsonPropertyName("outputTokens")] long OutputTokens,
+    [property: JsonPropertyName("costUsd")] decimal CostUsd);
+
+/// <summary>
+/// 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.
+/// </summary>
+public sealed record ExecutorPhaseRequest
+{
+    /// <summary>Work item the phase belongs to. Non-empty, at most 128 chars.</summary>
+    public required string WorkItemId { get; init; }
+
+    /// <summary>
+    /// Phase name (for example "work", "audit", "merge"). Open vocabulary so
+    /// future pipeline phases need no contract change; restricted to
+    /// <c>[A-Za-z0-9_-]</c>, at most 64 chars.
+    /// </summary>
+    public required string Phase { get; init; }
+
+    /// <summary>
+    /// Attempt number within the phase. Must be zero or positive; redelivery
+    /// of the same attempt is idempotent, a new attempt is a new dispatch.
+    /// </summary>
+    public required int Attempt { get; init; }
+
+    /// <summary>
+    /// Bare-repo id to stage to the executor (normally the work item id).
+    /// Only this repo is transferred — never the whole repos root.
+    /// </summary>
+    public required string RepositoryId { get; init; }
+
+    /// <summary>
+    /// Serialized phase input (for example the work item snapshot). Bounded
+    /// by dispatch options; covered by the idempotency body hash.
+    /// </summary>
+    public required string PayloadJson { get; init; }
+}
+
+/// <summary>
+/// Result of one phase execution: agent-visible outcome plus the commit the
+/// phase produced, the findings it reported, and the usage it consumed.
+/// </summary>
+public sealed record ExecutorPhaseResult
+{
+    public required ExecutorPhaseOutcome Outcome { get; init; }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    public string? CommitSha { get; init; }
+
+    /// <summary>Findings reported by the phase (for example audit findings).</summary>
+    public IReadOnlyList<string> Findings { get; init; } = [];
+
+    public required ExecutorPhaseUsage Usage { get; init; }
+
+    /// <summary>Agent-facing error detail for <see cref="ExecutorPhaseOutcome.AgentFailed"/>.</summary>
+    public string? ErrorMessage { get; init; }
+}

src/CodeyBox.Core/ExecutorPhaseTransport.cs

@@ -0,0 +1,131 @@
+namespace CodeyBox.Core;
+
+/// <summary>
+/// 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.
+/// </summary>
+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; }
+}
+
+/// <summary>
+/// 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 <see cref="ExecutorPhaseTransportException"/>:
+/// 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.
+/// </summary>
+public sealed class ExecutorPhaseException : Exception
+{
+    public ExecutorPhaseException(string message)
+        : base(message)
+    {
+    }
+
+    public ExecutorPhaseException(string message, Exception inner)
+        : base(message, inner)
+    {
+    }
+}
+
+/// <summary>
+/// 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.
+/// </summary>
+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; }
+}
+
+/// <summary>
+/// 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.
+///
+/// <para>All methods throw <see cref="ExecutorPhaseTransportException"/> 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.</para>
+/// </summary>
+public interface IExecutorPhaseTransport
+{
+    /// <summary>Stable executor host id this transport talks to.</summary>
+    string HostId { get; }
+
+    /// <summary>
+    /// Copies the host-local bare repo at <paramref name="hostRepoPath"/> to
+    /// the executor. Called with exactly one per-item repo path per dispatch.
+    /// </summary>
+    Task StageInAsync(string hostRepoPath, CancellationToken ct);
+
+    /// <summary>
+    /// 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 <see cref="ExecutorPhaseOutcome.AgentFailed"/>, not
+    /// thrown — only transport failures throw.
+    /// </summary>
+    Task<ExecutorPhaseResult> RunPhaseAsync(ExecutorPhaseRequest request, CancellationToken ct);
+
+    /// <summary>
+    /// Writes the executor-side repo back to a host-local tar archive at
+    /// <paramref name="hostArchivePath"/>. The caller validates the archive
+    /// (size, entry count, expansion ratio, path containment) before anything
+    /// is extracted over the orchestrator's bare repo.
+    ///
+    /// <para>The transport MUST enforce <paramref name="maxArchiveBytes"/>
+    /// 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
+    /// <see cref="ExecutorPhaseException"/> (a phase failure: the host was
+    /// reachable, the payload was hostile), never a transport exception, and
+    /// must leave no usable archive behind at
+    /// <paramref name="hostArchivePath"/>.</para>
+    /// </summary>
+    Task StageOutToArchiveAsync(string hostArchivePath, long maxArchiveBytes, CancellationToken ct);
+}
+
+/// <summary>
+/// Resolves the dispatch transport for a registered executor host. Returns
+/// null when the host has no transport configured (for example no SSH target
+/// is mapped for it); the proxy treats that as a transport failure rather
+/// than silently running the phase elsewhere.
+/// </summary>
+public interface IExecutorPhaseTransportFactory
+{
+    Task<IExecutorPhaseTransport?> ResolveAsync(string hostId, CancellationToken ct);
+}

src/CodeyBox.Orchestrator/ExecutorPhaseDispatchOptions.cs

@@ -0,0 +1,84 @@
+namespace CodeyBox.Orchestrator;
+
+/// <summary>
+/// Tuning knobs for executor phase dispatch (see <see cref="ExecutorPhaseProxy"/>).
+/// Bound under <c>CodeyBox:ExecutorPhaseDispatch</c>. 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.
+/// </summary>
+public sealed class ExecutorPhaseDispatchOptions
+{
+    /// <summary>
+    /// Maximum tar bytes accepted back from an executor per dispatch. The
+    /// transport enforces this cap while receiving (see
+    /// <c>IExecutorPhaseTransport.StageOutToArchiveAsync</c>) so
+    /// executor-controlled content cannot fill the orchestrator disk before
+    /// validation; the validator re-checks the landed size as defense in
+    /// depth.
+    /// Equivalent to <c>MultipassRemoteSandboxOptions.StageOutMaxArchiveBytes</c>.
+    /// </summary>
+    public long StageOutMaxArchiveBytes { get; set; } = 2L * 1024 * 1024 * 1024;
+
+    /// <summary>
+    /// Maximum non-metadata tar entries accepted per staged-back archive.
+    /// Equivalent to <c>MultipassRemoteSandboxOptions.StageOutMaxEntries</c>.
+    /// </summary>
+    public int StageOutMaxEntries { get; set; } = 200_000;
+
+    /// <summary>
+    /// Maximum declared regular-file payload divided by archive bytes.
+    /// Equivalent to <c>MultipassRemoteSandboxOptions.StageOutMaxExpansionRatio</c>.
+    /// </summary>
+    public double StageOutMaxExpansionRatio { get; set; } = 1.5d;
+
+    /// <summary>
+    /// How long a delivered dispatch result is replayed from the idempotency
+    /// store on redelivery. Mirrors the API idempotency TTL.
+    /// </summary>
+    public TimeSpan IdempotencyTtl { get; set; } = TimeSpan.FromHours(24);
+
+    /// <summary>Maximum serialized bytes accepted in a dispatch request payload.</summary>
+    public int MaxRequestPayloadBytes { get; set; } = 1024 * 1024;
+
+    /// <summary>Maximum findings accepted in an executor-returned result.</summary>
+    public int MaxResultFindings { get; set; } = 128;
+
+    /// <summary>Maximum chars accepted per finding in an executor-returned result.</summary>
+    public int MaxFindingLengthChars { get; set; } = 8192;
+
+    /// <summary>Maximum chars accepted in an executor-returned error message.</summary>
+    public int MaxResultErrorLengthChars { get; set; } = 8192;
+
+    /// <summary>
+    /// Fails fast on misconfiguration so a bad bound surfaces at dispatch
+    /// time instead of silently admitting an unbounded payload.
+    /// </summary>
+    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.");
+    }
+}