From 1c726fbb7798f7f08e85d97406d02ce14cbb572e Mon Sep 17 00:00:00 2001 From: Adam Frisby Date: Mon, 14 Sep 2026 11:38:24 +0000 Subject: [PATCH] Place executor phases by credential, network profile, capacity and capability Replace first-registered-wins selection with the pure ExecutorPlacement decider: credential (exact), network profile, required capabilities in the router vocabulary, capacity, cordon/health plus runtime backoff. Transient no-eligible-host defers under PlacementRecheckIn; unknown capabilities report unplaceable naming the tag; transport failures fail over across eligible hosts with host attribution. CodeyBox-Prompt-Revision: 1 Co-Authored-By: CodeyBox --- docs/operating/remote-executors.md | 39 ++- docs/reference/api.md | 3 + src/CodeyBox.Api/ExecutorEndpoints.cs | 4 + src/CodeyBox.Api/WorkerRegistryEndpoints.cs | 1 + src/CodeyBox.Core/ExecutorEligibility.cs | 73 +++++ src/CodeyBox.Core/ExecutorPhase.cs | 27 ++ src/CodeyBox.Core/ExecutorPlacement.cs | 273 ++++++++++++++++++ .../ExecutorPlacementUnplaceableException.cs | 43 +++ src/CodeyBox.Core/ExecutorRegistration.cs | 12 + src/CodeyBox.Core/WorkerRegistration.cs | 6 + src/CodeyBox.Orchestrator/ExecutorClient.cs | 1 + src/CodeyBox.Orchestrator/ExecutorOptions.cs | 9 + .../ExecutorPhaseDispatchOptions.cs | 23 ++ .../ExecutorPhaseProxy.cs | 255 +++++++++++++++- .../SqliteWorkerRegistry.cs | 19 +- .../ExecutorEligibilityTests.cs | 44 ++- tests/CodeyBox.Tests/ExecutorHostTests.cs | 31 ++ .../CodeyBox.Tests/ExecutorPhaseProxyTests.cs | 271 ++++++++++++++++- .../CodeyBox.Tests/ExecutorPlacementTests.cs | 204 +++++++++++++ 19 files changed, 1297 insertions(+), 41 deletions(-) create mode 100644 src/CodeyBox.Core/ExecutorPlacement.cs create mode 100644 src/CodeyBox.Core/ExecutorPlacementUnplaceableException.cs create mode 100644 tests/CodeyBox.Tests/ExecutorPlacementTests.cs diff --git a/docs/operating/remote-executors.md b/docs/operating/remote-executors.md index 5071f47b..62be6b62 100644 --- a/docs/operating/remote-executors.md +++ b/docs/operating/remote-executors.md @@ -13,9 +13,17 @@ 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 +implements `IExecutorPhaseRunner`: it places each phase on a registered +executor through the pure `ExecutorPlacement` decider +(`src/CodeyBox.Core/ExecutorPlacement.cs`), matching the phase's requirements +against each host's declared attributes — the agent credential the route +needs against `DeclaredCredentials` (exact equality), the sandbox target's +network profile against `AllowedNetworkProfiles` (empty means all), and the +work item's `RequiredCapabilities` against the host's `DeclaredCapabilities` +in the same case-insensitive capability vocabulary the agent-class router +uses. Cordoned, unhealthy, runtime-backed-off and at-capacity hosts are +excluded; among the eligible hosts the least-loaded wins (ties break by host +id). The proxy then 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 @@ -27,7 +35,8 @@ 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 +phase + attempt and the body hash covers the request (including the placement +requirements when set), 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 @@ -35,15 +44,30 @@ 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 +and stores nothing, so an unreachable host fails over to the next eligible +host (and, when every eligible host fails, the last host-attributed failure +propagates) rather than being charged against the work item as an agent +failure. A host that declared a credential it does not actually hold surfaces +the same way — as a host-attributed failure with failover — never as an agent +failure. When hosts are registered but none is currently eligible, the +dispatch is deferred under `PlacementRecheckIn` so the work item is requeued +rather than failed; when no registered host provides a required capability, +the item is reported unplaceable naming the unmet tag instead of being +dispatched and failed, and neither path consumes a rework iteration. Every +decision is logged with the chosen host and the per-candidate refusal reason. +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. +`MaxResultErrorLengthChars`, `PlacementRecheckIn`, `RuntimeUnhealthyBackoff`), +hot-reloadable like the other dispatch knobs. `PlacementRecheckIn` (default +15 s, mirroring the remote sandbox provider) is the requeue delay used when +every eligible host is full, cordoned or unhealthy; `RuntimeUnhealthyBackoff` +(default 1 min) is how long a host that fails dispatch is skipped before the +next dispatch probes it again. ## Running the executor @@ -65,6 +89,7 @@ All operational values live under `CodeyBox:Executor` and are hot-reloadable | `MaxConcurrentSandboxes` | `int?` | `null` (uncapped) | Host-local sandbox capacity. `0` registers but is never selected | | `AllowedNetworkProfiles` | `string[]` | `[]` (all) | Network profiles this host accepts; `"*"` also means all | | `DeclaredCredentials` | `string[]` | `[]` | Agent credential sets this host holds (e.g. `claude`, `codex`) | +| `DeclaredCapabilities` | `string[]` | `[]` | Clearance tags this host may handle, in the work item `RequiredCapabilities` vocabulary | | `Cordoned` | `bool` | `false` | Draining: registers and heartbeats but is never selected | | `Healthy` | `bool` | `true` | Health gate: `false` routes placements away without unregistering | | `LocalSandboxProvider` | `string` | `process` | `process` (dev runner, UNSAFE) or `bubblewrap` | diff --git a/docs/reference/api.md b/docs/reference/api.md index 984c7107..66772dd0 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -1330,6 +1330,7 @@ Response: `200 OK` with a JSON array: | `maxConcurrentSandboxes` | Declared host-local sandbox capacity (`null` = uncapped / non-executor row) | | `executorNetworkProfiles` | Declared network profiles the executor accepts (empty = all; `null` for non-executor rows) | | `executorCredentials` | Names of the agent credential sets the executor holds (`null` for non-executor rows) | +| `executorCapabilities` | Clearance tags the executor declares, in the work item `RequiredCapabilities` vocabulary (`null` for non-executor rows) | | `cordoned` | Draining flag: registers and heartbeats but is never selected for new placements | | `healthy` | Operator health gate: `false` routes new placements away without removing the registration | @@ -1347,6 +1348,7 @@ Request (`application/json`): "maxConcurrentSandboxes": 2, "allowedNetworkProfiles": ["restricted"], "declaredCredentials": ["claude"], + "declaredCapabilities": ["sensitive"], "cordoned": false, "healthy": true, "processId": 12345 @@ -1359,6 +1361,7 @@ Request (`application/json`): | `maxConcurrentSandboxes` | Host-local sandbox capacity, 0–100000. `0` registers the executor but leaves it never selected; omit for uncapped | | `allowedNetworkProfiles` | At most 64 entries; empty (or `"*"`) accepts every profile | | `declaredCredentials` | At most 64 opaque credential-set names, matched by exact equality | +| `declaredCapabilities` | At most 64 clearance tags in the work item `RequiredCapabilities` vocabulary, matched case-insensitively | | `cordoned` | Draining flag: registers and heartbeats but is never selected for new placements | | `healthy` | Health gate, default `true` | | `processId` | Executor OS process id, informational only | diff --git a/src/CodeyBox.Api/ExecutorEndpoints.cs b/src/CodeyBox.Api/ExecutorEndpoints.cs index 3ae2d15e..e190ea5f 100644 --- a/src/CodeyBox.Api/ExecutorEndpoints.cs +++ b/src/CodeyBox.Api/ExecutorEndpoints.cs @@ -52,10 +52,12 @@ private static async Task RegisterAsync( string[] profiles; string[] credentials; + string[] capabilities; try { profiles = NormalizeEntries(req.AllowedNetworkProfiles, nameof(req.AllowedNetworkProfiles)); credentials = NormalizeEntries(req.DeclaredCredentials, nameof(req.DeclaredCredentials)); + capabilities = NormalizeEntries(req.DeclaredCapabilities, nameof(req.DeclaredCapabilities)); } catch (ArgumentException ex) { @@ -74,6 +76,7 @@ private static async Task RegisterAsync( MaxConcurrentSandboxes = req.MaxConcurrentSandboxes, ExecutorNetworkProfiles = profiles, ExecutorCredentials = credentials, + ExecutorCapabilities = capabilities, Cordoned = req.Cordoned, Healthy = req.Healthy ?? true, }; @@ -191,6 +194,7 @@ public sealed class ExecutorRegistrationRequest public int? MaxConcurrentSandboxes { get; set; } public List? AllowedNetworkProfiles { get; set; } public List? DeclaredCredentials { get; set; } + public List? DeclaredCapabilities { get; set; } public bool Cordoned { get; set; } public bool? Healthy { get; set; } public int? ProcessId { get; set; } diff --git a/src/CodeyBox.Api/WorkerRegistryEndpoints.cs b/src/CodeyBox.Api/WorkerRegistryEndpoints.cs index 19b4fdb8..bc0692b6 100644 --- a/src/CodeyBox.Api/WorkerRegistryEndpoints.cs +++ b/src/CodeyBox.Api/WorkerRegistryEndpoints.cs @@ -28,6 +28,7 @@ private static async Task ListWorkersAsync(IWorkerRegistry registry, Ca maxConcurrentSandboxes = w.MaxConcurrentSandboxes, executorNetworkProfiles = w.ExecutorNetworkProfiles, executorCredentials = w.ExecutorCredentials, + executorCapabilities = w.ExecutorCapabilities, cordoned = w.Cordoned, healthy = w.Healthy, })); diff --git a/src/CodeyBox.Core/ExecutorEligibility.cs b/src/CodeyBox.Core/ExecutorEligibility.cs index 0b216b7f..77a2a891 100644 --- a/src/CodeyBox.Core/ExecutorEligibility.cs +++ b/src/CodeyBox.Core/ExecutorEligibility.cs @@ -75,4 +75,77 @@ public static bool HoldsCredential(ExecutorRegistration registration, string cre } return false; } + + /// + /// True when the executor's declared capabilities cover every required + /// tag. Uses the same vocabulary and comparison as the agent-class + /// router's RequiredCapabilities gate: ordinal, case-insensitive, + /// exact equality per tag. An empty required set is covered by any host. + /// + public static bool CoversRequiredCapabilities( + ExecutorRegistration registration, + IReadOnlyList? required) + { + ArgumentNullException.ThrowIfNull(registration); + if (required is null || required.Count == 0) + return true; + if (registration.DeclaredCapabilities.Count == 0) + return false; + foreach (var tag in required) + { + if (string.IsNullOrWhiteSpace(tag)) + continue; + var wanted = tag.Trim(); + var hit = false; + foreach (var have in registration.DeclaredCapabilities) + { + if (string.Equals(have?.Trim(), wanted, StringComparison.OrdinalIgnoreCase)) + { + hit = true; + break; + } + } + if (!hit) + return false; + } + return true; + } + + /// + /// First required capability no host in declares + /// (ordinal, case-insensitive), or null when every required tag is held + /// by at least one host. Blank required entries are ignored. + /// + public static string? FindCapabilityNoHostProvides( + IEnumerable hosts, + IReadOnlyList? required) + { + ArgumentNullException.ThrowIfNull(hosts); + if (required is null || required.Count == 0) + return null; + foreach (var tag in required) + { + if (string.IsNullOrWhiteSpace(tag)) + continue; + var wanted = tag.Trim(); + var provided = false; + foreach (var host in hosts) + { + ArgumentNullException.ThrowIfNull(host); + foreach (var have in host.DeclaredCapabilities) + { + if (string.Equals(have?.Trim(), wanted, StringComparison.OrdinalIgnoreCase)) + { + provided = true; + break; + } + } + if (provided) + break; + } + if (!provided) + return wanted; + } + return null; + } } diff --git a/src/CodeyBox.Core/ExecutorPhase.cs b/src/CodeyBox.Core/ExecutorPhase.cs index 5090574e..49b81326 100644 --- a/src/CodeyBox.Core/ExecutorPhase.cs +++ b/src/CodeyBox.Core/ExecutorPhase.cs @@ -61,6 +61,33 @@ public sealed record ExecutorPhaseRequest /// by dispatch options; covered by the idempotency body hash. /// public required string PayloadJson { get; init; } + + /// + /// Agent credential set the phase's route requires (for example "claude"). + /// Null or blank means the phase needs no specific credential and any + /// host may run it. Matched against + /// by exact + /// ordinal equality. At most 128 chars. + /// + public string? RequiredCredential { get; init; } + + /// + /// Sandbox network profile the phase's sandbox target requires. Null or + /// blank means "(default)". Matched against + /// with the + /// same empty-means-all / "*" semantics. At most 128 chars. + /// + public string? RequiredNetworkProfile { get; init; } + + /// + /// Clearance tags the phase's work item demands, in the same vocabulary + /// as . Empty means no + /// clearance required. Placement only selects hosts whose + /// covers every + /// tag here (ordinal, case-insensitive). At most 16 entries, each at most + /// 128 chars. + /// + public IReadOnlyList RequiredCapabilities { get; init; } = []; } /// diff --git a/src/CodeyBox.Core/ExecutorPlacement.cs b/src/CodeyBox.Core/ExecutorPlacement.cs new file mode 100644 index 00000000..bf71d9b5 --- /dev/null +++ b/src/CodeyBox.Core/ExecutorPlacement.cs @@ -0,0 +1,273 @@ +namespace CodeyBox.Core; + +/// +/// Placement requirements for one phase dispatch. Derived from the +/// placement fields so the pure +/// decider stays decoupled from the +/// dispatch envelope. +/// +public sealed record ExecutorPlacementRequirements +{ + /// Maximum entries accepted in . + public const int MaxRequiredCapabilities = 16; + + /// Maximum chars accepted in a credential, profile or capability entry. + public const int MaxEntryLength = 128; + + /// Credential the route needs, or null when the phase needs none. + public string? RequiredCredential { get; init; } + + /// Network profile the sandbox target requires, or null for "(default)". + public string? RequiredNetworkProfile { get; init; } + + /// Clearance tags the work item demands, in the existing capability vocabulary. + public IReadOnlyList RequiredCapabilities { get; init; } = []; + + /// + /// Builds requirements from a dispatch request. Trims entries; blank + /// credential/profile become null; blank capability entries are dropped. + /// Throws when a bound is exceeded so + /// misconfigured callers fail fast instead of silently widening placement. + /// + public static ExecutorPlacementRequirements FromRequest(ExecutorPhaseRequest request) + { + ArgumentNullException.ThrowIfNull(request); + var credential = string.IsNullOrWhiteSpace(request.RequiredCredential) + ? null + : request.RequiredCredential.Trim(); + if (credential is not null && credential.Length > MaxEntryLength) + throw new ArgumentException( + $"RequiredCredential must be at most {MaxEntryLength} characters.", nameof(request)); + if (credential is not null && credential.Any(char.IsControl)) + throw new ArgumentException("RequiredCredential must not contain control characters.", nameof(request)); + + var profile = string.IsNullOrWhiteSpace(request.RequiredNetworkProfile) + ? null + : request.RequiredNetworkProfile.Trim(); + if (profile is not null && profile.Length > MaxEntryLength) + throw new ArgumentException( + $"RequiredNetworkProfile must be at most {MaxEntryLength} characters.", nameof(request)); + if (profile is not null && profile.Any(char.IsControl)) + throw new ArgumentException("RequiredNetworkProfile must not contain control characters.", nameof(request)); + + var capabilities = request.RequiredCapabilities ?? []; + if (capabilities.Count > MaxRequiredCapabilities) + throw new ArgumentException( + $"RequiredCapabilities may contain at most {MaxRequiredCapabilities} entries.", nameof(request)); + var normalised = new List(capabilities.Count); + foreach (var raw in capabilities) + { + if (string.IsNullOrWhiteSpace(raw)) + continue; + var tag = raw.Trim(); + if (tag.Length > MaxEntryLength) + throw new ArgumentException( + $"RequiredCapabilities entries must be at most {MaxEntryLength} characters.", nameof(request)); + if (tag.Any(char.IsControl)) + throw new ArgumentException("RequiredCapabilities entries must not contain control characters.", nameof(request)); + normalised.Add(tag); + } + + return new ExecutorPlacementRequirements + { + RequiredCredential = credential, + RequiredNetworkProfile = profile, + RequiredCapabilities = normalised, + }; + } +} + +/// Per-candidate outcome of a placement decision, for observability. +public sealed record ExecutorPlacementCandidateOutcome +{ + public required string HostId { get; init; } + + /// True when this host may receive the phase. + public required bool Eligible { get; init; } + + /// + /// Machine-readable exclusion reason for ineligible hosts, or "eligible" + /// / "selected" for eligible ones. Values: selected, + /// eligible, cordoned, unhealthy, + /// runtime-unhealthy, at-capacity, + /// missing-credential:<name>, + /// network-profile:<profile>, + /// missing-capability:<tag>, not-selected. + /// + public required string Reason { get; init; } +} + +/// +/// Observable result of a placement decision: the chosen host (if any) plus +/// the per-candidate reason list. When no host is eligible, +/// names the required tag no registered host +/// provides (permanent, unplaceable); null means the refusal is transient +/// (capacity, cordon, health, credential or profile mismatch on the +/// currently-available set) and the caller must requeue under backoff. +/// +public sealed record ExecutorPlacementDecision +{ + public required string? SelectedHostId { get; init; } + + public required IReadOnlyList Candidates { get; init; } + + /// Required capability no registered host declares, if any. + public string? UnmetCapability { get; init; } + + /// True when the item can never place until registration changes. + public bool IsUnplaceable => UnmetCapability is not null; + + /// Human-readable one-line summary for logs and exception details. + public string Describe() + { + if (SelectedHostId is not null) + return $"selected={SelectedHostId}; candidates=[{string.Join(", ", Candidates.Select(c => $"{c.HostId}={c.Reason}"))}]"; + if (UnmetCapability is not null) + return $"unplaceable missing-capability={UnmetCapability}; candidates=[{string.Join(", ", Candidates.Select(c => $"{c.HostId}={c.Reason}"))}]"; + return $"no-eligible-host; candidates=[{string.Join(", ", Candidates.Select(c => $"{c.HostId}={c.Reason}"))}]"; + } +} + +/// +/// Pure executor-host placement decider. Matches a phase's requirements +/// (agent credential, network profile, required capabilities) against each +/// host's declared attributes, excluding cordoned and unhealthy hosts and +/// hosts at capacity. Deterministic: among eligible hosts the least-loaded +/// wins, ties broken by fewest in-flight reservations then ordinal host id — +/// mirroring the multipass-remote sandbox placement ordering so the two +/// placement paths cannot drift apart. +/// +public static class ExecutorPlacement +{ + /// + /// Decides placement over the given hosts. + /// carries the current live load per host id (missing means zero); + /// lists host ids under runtime + /// backoff (skipped like unhealthy hosts). Never throws for empty input: + /// with no registered hosts the decision simply selects nothing. + /// + public static ExecutorPlacementDecision Decide( + IReadOnlyList hosts, + ExecutorPlacementRequirements requirements, + IReadOnlyDictionary? loads = null, + ISet? runtimeUnhealthy = null) + { + ArgumentNullException.ThrowIfNull(hosts); + ArgumentNullException.ThrowIfNull(requirements); + + var outcomes = new List(hosts.Count); + string? selected = null; + var selectedLoad = double.MaxValue; + var selectedUsed = int.MaxValue; + + foreach (var host in hosts.OrderBy(h => h.HostId, StringComparer.Ordinal)) + { + var used = 0; + if (loads is not null && loads.TryGetValue(host.HostId, out var load)) + used = Math.Max(0, load); + var reason = ExcludeReason(host, requirements, used, runtimeUnhealthy); + if (reason is null) + { + var capacity = host.MaxConcurrentSandboxes is { } cap ? cap : int.MaxValue; + var loadRatio = capacity == int.MaxValue ? 0.0d : (double)used / capacity; + outcomes.Add(new ExecutorPlacementCandidateOutcome { HostId = host.HostId, Eligible = true, Reason = "eligible" }); + if (selected is null + || loadRatio < selectedLoad + || (Math.Abs(loadRatio - selectedLoad) < double.Epsilon && used < selectedUsed)) + { + selected = host.HostId; + selectedLoad = loadRatio; + selectedUsed = used; + } + } + else + { + outcomes.Add(new ExecutorPlacementCandidateOutcome { HostId = host.HostId, Eligible = false, Reason = reason }); + } + } + + for (var i = 0; i < outcomes.Count; i++) + { + if (selected is not null + && outcomes[i].Eligible + && string.Equals(outcomes[i].HostId, selected, StringComparison.Ordinal)) + { + outcomes[i] = outcomes[i] with { Reason = "selected" }; + } + else if (outcomes[i].Eligible) + { + outcomes[i] = outcomes[i] with { Reason = "not-selected" }; + } + } + + string? unmet = null; + if (selected is null && hosts.Count > 0) + unmet = ExecutorEligibility.FindCapabilityNoHostProvides(hosts, requirements.RequiredCapabilities); + + return new ExecutorPlacementDecision + { + SelectedHostId = selected, + Candidates = outcomes, + UnmetCapability = unmet, + }; + } + + private static string? ExcludeReason( + ExecutorRegistration host, + ExecutorPlacementRequirements requirements, + int used, + ISet? runtimeUnhealthy) + { + if (host.Cordoned) + return "cordoned"; + if (!host.Healthy) + return "unhealthy"; + if (runtimeUnhealthy is not null && runtimeUnhealthy.Contains(host.HostId)) + return "runtime-unhealthy"; + if (host.MaxConcurrentSandboxes is { } capacity) + { + if (capacity <= 0) + return $"at-capacity({used}/0)"; + if (used >= capacity) + return $"at-capacity({used}/{capacity})"; + } + if (!string.IsNullOrEmpty(requirements.RequiredCredential) + && !ExecutorEligibility.HoldsCredential(host, requirements.RequiredCredential!)) + return $"missing-credential:{requirements.RequiredCredential}"; + var profile = string.IsNullOrWhiteSpace(requirements.RequiredNetworkProfile) + ? null + : requirements.RequiredNetworkProfile.Trim(); + if (!ExecutorEligibility.AcceptsNetworkProfile(host, profile)) + return $"network-profile:{(profile ?? "(default)")}"; + if (!ExecutorEligibility.CoversRequiredCapabilities(host, requirements.RequiredCapabilities)) + { + var missing = FirstMissingCapability(host, requirements.RequiredCapabilities); + return $"missing-capability:{missing}"; + } + return null; + } + + private static string FirstMissingCapability( + ExecutorRegistration host, + IReadOnlyList required) + { + foreach (var tag in required) + { + if (string.IsNullOrWhiteSpace(tag)) + continue; + var wanted = tag.Trim(); + var hit = false; + foreach (var have in host.DeclaredCapabilities) + { + if (string.Equals(have?.Trim(), wanted, StringComparison.OrdinalIgnoreCase)) + { + hit = true; + break; + } + } + if (!hit) + return wanted; + } + return "(unknown)"; + } +} diff --git a/src/CodeyBox.Core/ExecutorPlacementUnplaceableException.cs b/src/CodeyBox.Core/ExecutorPlacementUnplaceableException.cs new file mode 100644 index 00000000..e59d5897 --- /dev/null +++ b/src/CodeyBox.Core/ExecutorPlacementUnplaceableException.cs @@ -0,0 +1,43 @@ +namespace CodeyBox.Core; + +/// +/// A phase cannot place because its work item demands a capability no +/// registered executor host provides. Permanent until registration changes: +/// the caller must report the item as unplaceable naming +/// — not dispatch it, not fail it as an agent +/// failure, and not consume a rework iteration. Distinct from +/// (transient: capacity, +/// cordon, health — requeue under backoff) and from +/// (host failure — retry +/// elsewhere) so each outcome keeps its own recovery path. +/// +public sealed class ExecutorPlacementUnplaceableException : Exception +{ + public ExecutorPlacementUnplaceableException( + string unmetCapability, + string detail, + ExecutorPlacementDecision? decision = null, + Exception? innerException = null) + : base(BuildMessage(unmetCapability, detail), innerException) + { + ArgumentException.ThrowIfNullOrWhiteSpace(unmetCapability); + UnmetCapability = unmetCapability.Trim(); + Detail = detail; + Decision = decision; + } + + /// Required capability tag no registered host declares. + public string UnmetCapability { get; } + + /// Per-candidate placement detail for logs; carries host ids and reasons only. + public string Detail { get; } + + /// Full placement decision, when the caller computed one. + public ExecutorPlacementDecision? Decision { get; } + + private static string BuildMessage(string unmetCapability, string detail) + { + var suffix = string.IsNullOrWhiteSpace(detail) ? "" : $": {detail.Trim()}"; + return $"executor placement unplaceable: no registered host provides capability '{unmetCapability.Trim()}'{suffix}"; + } +} diff --git a/src/CodeyBox.Core/ExecutorRegistration.cs b/src/CodeyBox.Core/ExecutorRegistration.cs index f18cad4c..ce79ccd9 100644 --- a/src/CodeyBox.Core/ExecutorRegistration.cs +++ b/src/CodeyBox.Core/ExecutorRegistration.cs @@ -62,6 +62,18 @@ public sealed record ExecutorRegistration /// public IReadOnlyList DeclaredCredentials { get; init; } = []; + /// + /// Clearance tags this host is trusted to handle, in the same vocabulary + /// as and + /// (for example "sensitive", + /// "audit"). A phase whose work item demands capabilities is only placed + /// on a host covering every required tag. Tag comparison is ordinal, + /// case-insensitive — the same comparison the agent-class router uses — + /// so hosts extend the existing capability vocabulary instead of + /// introducing a parallel one. + /// + public IReadOnlyList DeclaredCapabilities { get; init; } = []; + /// /// When true the host is draining: it registers and heartbeats but is /// never selected for new placements, mirroring diff --git a/src/CodeyBox.Core/WorkerRegistration.cs b/src/CodeyBox.Core/WorkerRegistration.cs index c2e1bcb8..1c73905f 100644 --- a/src/CodeyBox.Core/WorkerRegistration.cs +++ b/src/CodeyBox.Core/WorkerRegistration.cs @@ -48,6 +48,12 @@ public sealed record WorkerRegistration /// public IReadOnlyList? ExecutorCredentials { get; init; } + /// + /// Clearance tags the executor declares, in the same vocabulary as + /// . Null for non-executor rows. + /// + public IReadOnlyList? ExecutorCapabilities { get; init; } + /// /// Draining flag from the executor's registration. True means the host /// registers and heartbeats but is never selected for new placements. diff --git a/src/CodeyBox.Orchestrator/ExecutorClient.cs b/src/CodeyBox.Orchestrator/ExecutorClient.cs index 7bd06a1c..96494b4f 100644 --- a/src/CodeyBox.Orchestrator/ExecutorClient.cs +++ b/src/CodeyBox.Orchestrator/ExecutorClient.cs @@ -134,6 +134,7 @@ public async Task RegisterAsync(CancellationToken ct = default) maxConcurrentSandboxes = registration.MaxConcurrentSandboxes, allowedNetworkProfiles = registration.AllowedNetworkProfiles, declaredCredentials = registration.DeclaredCredentials, + declaredCapabilities = registration.DeclaredCapabilities, cordoned = registration.Cordoned, healthy = registration.Healthy, processId = Environment.ProcessId, diff --git a/src/CodeyBox.Orchestrator/ExecutorOptions.cs b/src/CodeyBox.Orchestrator/ExecutorOptions.cs index 8d08a9f4..2225141a 100644 --- a/src/CodeyBox.Orchestrator/ExecutorOptions.cs +++ b/src/CodeyBox.Orchestrator/ExecutorOptions.cs @@ -42,6 +42,13 @@ public sealed class ExecutorOptions /// Names of the agent credential sets this host holds. public List DeclaredCredentials { get; set; } = []; + /// + /// Clearance tags this host is trusted to handle, in the same vocabulary + /// as the work item RequiredCapabilities clearance tags. The + /// orchestrator only places phases demanding a tag on hosts declaring it. + /// + public List DeclaredCapabilities { get; set; } = []; + /// When true the host drains: registers and heartbeats but is never selected for new placements. public bool Cordoned { get; set; } @@ -80,6 +87,7 @@ public sealed class ExecutorOptions MaxConcurrentSandboxes = MaxConcurrentSandboxes, AllowedNetworkProfiles = [.. AllowedNetworkProfiles], DeclaredCredentials = [.. DeclaredCredentials], + DeclaredCapabilities = [.. DeclaredCapabilities], Cordoned = Cordoned, Healthy = Healthy, }; @@ -112,6 +120,7 @@ public void Validate() $"CodeyBox:Executor:MaxConcurrentSandboxes must be between 0 and {ExecutorRegistration.MaxDeclaredCapacity}."); ValidateEntries(AllowedNetworkProfiles, nameof(AllowedNetworkProfiles)); ValidateEntries(DeclaredCredentials, nameof(DeclaredCredentials)); + ValidateEntries(DeclaredCapabilities, nameof(DeclaredCapabilities)); if (HeartbeatInterval <= TimeSpan.Zero) throw new InvalidOperationException("CodeyBox:Executor:HeartbeatInterval must be positive."); if (RequestTimeout <= TimeSpan.Zero) diff --git a/src/CodeyBox.Orchestrator/ExecutorPhaseDispatchOptions.cs b/src/CodeyBox.Orchestrator/ExecutorPhaseDispatchOptions.cs index 8d9051fc..76a1b478 100644 --- a/src/CodeyBox.Orchestrator/ExecutorPhaseDispatchOptions.cs +++ b/src/CodeyBox.Orchestrator/ExecutorPhaseDispatchOptions.cs @@ -50,6 +50,23 @@ public sealed class ExecutorPhaseDispatchOptions /// Maximum chars accepted in an executor-returned error message. public int MaxResultErrorLengthChars { get; set; } = 8192; + /// + /// Requeue delay surfaced when no executor host can currently accept a + /// phase because every eligible host is full, cordoned, unhealthy, or + /// mismatched on credential/profile. Mirrors + /// MultipassRemoteSandboxOptions.PlacementRecheckIn: the work item + /// is requeued under this backoff rather than failed. Hot-reloadable. + /// + public TimeSpan PlacementRecheckIn { get; set; } = TimeSpan.FromSeconds(15); + + /// + /// How long a host that fails dispatch with a transport error is skipped + /// for new placements before the next dispatch probes it again. Mirrors + /// MultipassRemoteSandboxOptions.RuntimeUnhealthyBackoff. + /// Hot-reloadable. + /// + public TimeSpan RuntimeUnhealthyBackoff { get; set; } = TimeSpan.FromMinutes(1); + /// /// Fails fast on misconfiguration so a bad bound surfaces at dispatch /// time instead of silently admitting an unbounded payload. @@ -80,5 +97,11 @@ public void Validate() if (MaxResultErrorLengthChars <= 0) throw new InvalidOperationException( "CodeyBox:ExecutorPhaseDispatch:MaxResultErrorLengthChars must be > 0."); + if (PlacementRecheckIn <= TimeSpan.Zero) + throw new InvalidOperationException( + "CodeyBox:ExecutorPhaseDispatch:PlacementRecheckIn must be positive."); + if (RuntimeUnhealthyBackoff <= TimeSpan.Zero) + throw new InvalidOperationException( + "CodeyBox:ExecutorPhaseDispatch:RuntimeUnhealthyBackoff must be positive."); } } diff --git a/src/CodeyBox.Orchestrator/ExecutorPhaseProxy.cs b/src/CodeyBox.Orchestrator/ExecutorPhaseProxy.cs index 0cb5d11b..f94229b8 100644 --- a/src/CodeyBox.Orchestrator/ExecutorPhaseProxy.cs +++ b/src/CodeyBox.Orchestrator/ExecutorPhaseProxy.cs @@ -1,3 +1,4 @@ +using System.Collections.Concurrent; using System.Security.Cryptography; using System.Text; using System.Text.Json; @@ -21,12 +22,18 @@ namespace CodeyBox.Orchestrator; /// 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. +/// through the pure decider: the phase's +/// required agent credential, required network profile and required +/// capabilities are matched against each host's declared attributes, and +/// cordoned, unhealthy, runtime-backed-off and at-capacity hosts are +/// excluded. The decision — chosen host plus the per-candidate refusal +/// reason — is logged for observability. With no executor registered at all, +/// 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. +/// path is ever transferred — never the whole repos root. A transport +/// failure fails over to the next eligible host; only when every eligible +/// host fails does the last host-attributed failure propagate. /// 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. @@ -38,9 +45,19 @@ namespace CodeyBox.Orchestrator; /// 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. +/// work item as an agent failure. A host that declared a credential it does +/// not actually hold surfaces the same way — as a host-attributed transport +/// failure with failover to another eligible host — never as an agent +/// failure. When hosts are registered but none is currently eligible, the +/// dispatch throws with +/// the configured placement backoff so the work item is requeued rather than +/// failed; when no registered host provides a required capability, it throws +/// naming the unmet tag +/// instead of dispatching and failing. Neither path returns +/// , so neither consumes a +/// rework iteration. 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 { @@ -50,6 +67,22 @@ public sealed class ExecutorPhaseProxy : IExecutorPhaseRunner private const int MaxWorkItemIdLength = 128; private const int MaxRepositoryIdLength = 256; + /// + /// Prefix an executor-side handler uses to report that the host cannot + /// satisfy the dispatch's required credential (for example the credential + /// file is absent although the host declared it). An + /// result whose + /// ErrorMessage starts with this prefix (ordinal) is reinterpreted + /// by the proxy as a host-attributed transport failure: the host is + /// marked runtime-unhealthy and the phase fails over to the next eligible + /// host instead of being charged to the work item as an agent failure. + /// Executor-side code that detects a missing credential directly must + /// throw instead; the + /// prefix exists for transports that can only surface the condition as + /// result text. + /// + public const string CredentialMissingErrorPrefix = "executor-credential-missing:"; + private readonly IWorkerRegistry _registry; private readonly IExecutorPhaseTransportFactory _transports; private readonly IGitHost _gitHost; @@ -59,6 +92,10 @@ public sealed class ExecutorPhaseProxy : IExecutorPhaseRunner private readonly TimeProvider _clock; private readonly ILogger _log; + private readonly ConcurrentDictionary _inflightByHost = new(StringComparer.Ordinal); + private readonly Dictionary _runtimeUnhealthy = new(StringComparer.Ordinal); + private readonly object _runtimeUnhealthyLock = new(); + public ExecutorPhaseProxy( IWorkerRegistry registry, IExecutorPhaseTransportFactory transports, @@ -104,8 +141,8 @@ public async Task ExecutePhaseAsync(ExecutorPhaseRequest re throw new InvalidOperationException($"Unknown idempotency outcome {(int)lookup.Outcome}."); } - var hostId = await SelectExecutorAsync(ct).ConfigureAwait(false); - if (hostId is null) + var hostIds = await SelectExecutorChainAsync(request, options, ct).ConfigureAwait(false); + if (hostIds 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); @@ -116,13 +153,51 @@ await _idempotency.PutAsync( return validatedFallback; } - var result = await ExecuteRemoteAsync(request, hostId, options, ct).ConfigureAwait(false); + var result = await ExecuteRemoteWithFailoverAsync(request, hostIds, options, dispatchKey, 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 ExecuteRemoteWithFailoverAsync( + ExecutorPhaseRequest request, + IReadOnlyList hostIds, + ExecutorPhaseDispatchOptions options, + string dispatchKey, + CancellationToken ct) + { + ExecutorPhaseTransportException? lastTransport = null; + foreach (var hostId in hostIds) + { + ct.ThrowIfCancellationRequested(); + TrackDispatchStart(hostId); + try + { + var result = await ExecuteRemoteAsync(request, hostId, options, ct).ConfigureAwait(false); + return result; + } + catch (ExecutorPhaseTransportException ex) + { + lastTransport = ex; + MarkRuntimeUnhealthy(ex.HostId, ex.Message, options.RuntimeUnhealthyBackoff); + if (hostIds.Count > 1) + { + _log.LogWarning( + "Executor phase dispatch {DispatchKey} failed on host {HostId} ({Operation}); failing over to another eligible host", + dispatchKey, ex.HostId, ex.Operation); + } + } + finally + { + TrackDispatchEnd(hostId); + } + } + + throw lastTransport + ?? new ExecutorPhaseTransportException("(unknown)", "placement", "No eligible executor host was attempted."); + } + private async Task ExecuteRemoteAsync( ExecutorPhaseRequest request, string hostId, @@ -148,6 +223,7 @@ private async Task ExecuteRemoteAsync( var raw = await CallTransportAsync(hostId, "run-phase", token => transport.RunPhaseAsync(request, token), ct).ConfigureAwait(false); var result = ValidateResult(raw, options); + ThrowIfCredentialMissingResult(hostId, result, request); var scratchRoot = Path.Combine(Path.GetTempPath(), "codeybox-executor-phase-" + Guid.NewGuid().ToString("N")); Directory.CreateDirectory(scratchRoot); @@ -227,10 +303,39 @@ private static Task CallTransportAsync( CancellationToken ct) => CallTransportAsync(hostId, operation, async token => { await call(token).ConfigureAwait(false); return null; }, ct); - private async Task SelectExecutorAsync(CancellationToken ct) + private static void ThrowIfCredentialMissingResult( + string hostId, + ExecutorPhaseResult result, + ExecutorPhaseRequest request) + { + if (result.Outcome != ExecutorPhaseOutcome.AgentFailed) + return; + if (string.IsNullOrEmpty(result.ErrorMessage) + || !result.ErrorMessage.StartsWith(CredentialMissingErrorPrefix, StringComparison.Ordinal)) + return; + var credential = string.IsNullOrWhiteSpace(request.RequiredCredential) + ? "required" + : $"'{request.RequiredCredential.Trim()}'"; + throw new ExecutorPhaseTransportException( + hostId, + "run-phase", + $"Host declared credential {credential} but cannot satisfy it: {TruncateForLog(result.ErrorMessage)}"); + } + + private static string TruncateForLog(string value, int maxLength = 256) + { + if (value.Length <= maxLength) + return value; + return value[..maxLength] + "…"; + } + + private async Task?> SelectExecutorChainAsync( + ExecutorPhaseRequest request, + ExecutorPhaseDispatchOptions options, + CancellationToken ct) { var workers = await _registry.ListAsync(ct).ConfigureAwait(false); - return workers + var hosts = workers .Where(w => w.IsExecutor && w.ExecutorHostId is not null) .Select(w => new ExecutorRegistration { @@ -238,15 +343,122 @@ private static Task CallTransportAsync( MaxConcurrentSandboxes = w.MaxConcurrentSandboxes, AllowedNetworkProfiles = w.ExecutorNetworkProfiles ?? [], DeclaredCredentials = w.ExecutorCredentials ?? [], + DeclaredCapabilities = w.ExecutorCapabilities ?? [], Cordoned = w.Cordoned, Healthy = w.Healthy, }) - .Where(reg => ExecutorEligibility.IsEligibleForPlacement(reg, currentLoad: 0)) - .OrderBy(reg => reg.HostId, StringComparer.Ordinal) - .Select(reg => reg.HostId) - .FirstOrDefault(); + .ToList(); + + if (hosts.Count == 0) + return null; + + var requirements = ExecutorPlacementRequirements.FromRequest(request); + var loads = BuildLoads(workers); + var now = _clock.GetUtcNow(); + var backedOff = SnapshotRuntimeUnhealthy(now, pruneExpired: true); + var decision = ExecutorPlacement.Decide(hosts, requirements, loads, backedOff); + + if (decision.SelectedHostId is not null) + { + _log.LogInformation( + "Executor phase dispatch for work item {WorkItemId} phase {Phase} placed on host {HostId}: {Decision}", + request.WorkItemId, request.Phase, decision.SelectedHostId, decision.Describe()); + var chain = new List(capacity: decision.Candidates.Count); + chain.Add(decision.SelectedHostId); + foreach (var candidate in decision.Candidates) + { + if (candidate.Eligible + && !string.Equals(candidate.HostId, decision.SelectedHostId, StringComparison.Ordinal)) + chain.Add(candidate.HostId); + } + return chain; + } + + if (decision.UnmetCapability is not null) + { + _log.LogWarning( + "Executor phase dispatch for work item {WorkItemId} phase {Phase} is unplaceable: {Decision}", + request.WorkItemId, request.Phase, decision.Describe()); + throw new ExecutorPlacementUnplaceableException( + decision.UnmetCapability, + $"workItem={request.WorkItemId} phase={request.Phase}; candidates=[{string.Join(", ", decision.Candidates.Select(c => $"{c.HostId}={c.Reason}"))}]", + decision); + } + + _log.LogWarning( + "Executor phase dispatch for work item {WorkItemId} phase {Phase} deferred: {Decision}", + request.WorkItemId, request.Phase, decision.Describe()); + throw new SandboxProvisioningDeferredException( + provider: "executor", + operation: "placement", + errorClass: "no-eligible-host", + detail: $"workItem={request.WorkItemId} phase={request.Phase}; hosts=[{string.Join(", ", decision.Candidates.Select(c => $"{c.HostId}={c.Reason}"))}]", + recheckIn: options.PlacementRecheckIn); } + private IReadOnlyDictionary BuildLoads(IReadOnlyList workers) + { + var loads = new Dictionary(StringComparer.Ordinal); + foreach (var worker in workers) + { + if (!worker.IsExecutor || worker.ExecutorHostId is null) + continue; + var baseLoad = string.IsNullOrWhiteSpace(worker.CurrentWorkItemId) ? 0 : 1; + var inflight = _inflightByHost.TryGetValue(worker.ExecutorHostId, out var active) ? Math.Max(0, active) : 0; + loads[worker.ExecutorHostId] = baseLoad + inflight; + } + foreach (var (hostId, active) in _inflightByHost) + { + if (!loads.ContainsKey(hostId) && active > 0) + loads[hostId] = Math.Max(0, active); + } + return loads; + } + + private void TrackDispatchStart(string hostId) => + _inflightByHost.AddOrUpdate(hostId, 1, (_, current) => current + 1); + + private void TrackDispatchEnd(string hostId) + { + _inflightByHost.AddOrUpdate(hostId, 0, (_, current) => Math.Max(0, current - 1)); + if (_inflightByHost.TryGetValue(hostId, out var current) && current <= 0) + _inflightByHost.TryRemove(hostId, out _); + } + + private void MarkRuntimeUnhealthy(string hostId, string reason, TimeSpan backoff) + { + if (string.IsNullOrWhiteSpace(hostId)) + return; + var until = _clock.GetUtcNow() + (backoff > TimeSpan.Zero ? backoff : TimeSpan.FromMinutes(1)); + lock (_runtimeUnhealthyLock) + { + _runtimeUnhealthy[hostId] = new RuntimeUnhealthyState(until, reason); + } + _log.LogWarning( + "Executor host {HostId} marked runtime-unhealthy until {Until:O}: {Reason}", + hostId, until, TruncateForLog(reason)); + } + + private HashSet SnapshotRuntimeUnhealthy(DateTimeOffset now, bool pruneExpired) + { + lock (_runtimeUnhealthyLock) + { + if (pruneExpired) + { + foreach (var (hostId, state) in _runtimeUnhealthy.ToArray()) + { + if (state.Until <= now) + _runtimeUnhealthy.Remove(hostId); + } + } + return new HashSet( + _runtimeUnhealthy.Where(kv => kv.Value.Until > now).Select(kv => kv.Key), + StringComparer.Ordinal); + } + } + + private sealed record RuntimeUnhealthyState(DateTimeOffset Until, string Reason); + internal static string BuildDispatchKey(ExecutorPhaseRequest request) => $"executor-phase/v1/{request.WorkItemId}/{request.Phase}/{request.Attempt}"; @@ -266,6 +478,16 @@ static void WriteField(SHA256 sha, string value) WriteField(sha, request.Attempt.ToString(System.Globalization.CultureInfo.InvariantCulture)); WriteField(sha, request.RepositoryId); WriteField(sha, request.PayloadJson ?? string.Empty); + if (!string.IsNullOrWhiteSpace(request.RequiredCredential) + || !string.IsNullOrWhiteSpace(request.RequiredNetworkProfile) + || (request.RequiredCapabilities is { Count: > 0 })) + { + WriteField(sha, "placement/v1"); + WriteField(sha, request.RequiredCredential?.Trim() ?? string.Empty); + WriteField(sha, request.RequiredNetworkProfile?.Trim() ?? string.Empty); + foreach (var tag in request.RequiredCapabilities ?? []) + WriteField(sha, tag?.Trim() ?? string.Empty); + } sha.TransformFinalBlock([], 0, 0); return Convert.ToHexString(sha.Hash!).ToLowerInvariant(); } @@ -285,6 +507,7 @@ internal static void ValidateRequest(ExecutorPhaseRequest request, ExecutorPhase var payloadBytes = Encoding.UTF8.GetByteCount(request.PayloadJson ?? string.Empty); if (payloadBytes > options.MaxRequestPayloadBytes) throw new ArgumentException($"PayloadJson exceeds MaxRequestPayloadBytes={options.MaxRequestPayloadBytes}.", nameof(request)); + ExecutorPlacementRequirements.FromRequest(request); } internal static ExecutorPhaseResult ValidateResult(ExecutorPhaseResult result, ExecutorPhaseDispatchOptions options) diff --git a/src/CodeyBox.Orchestrator/SqliteWorkerRegistry.cs b/src/CodeyBox.Orchestrator/SqliteWorkerRegistry.cs index 09df9239..9043267f 100644 --- a/src/CodeyBox.Orchestrator/SqliteWorkerRegistry.cs +++ b/src/CodeyBox.Orchestrator/SqliteWorkerRegistry.cs @@ -84,6 +84,7 @@ CREATE TABLE IF NOT EXISTS worker_registry ( max_concurrent_sandboxes INTEGER, executor_network_profiles TEXT, executor_credentials TEXT, + executor_capabilities TEXT, cordoned INTEGER NOT NULL DEFAULT 0, healthy INTEGER NOT NULL DEFAULT 1 ); @@ -114,8 +115,8 @@ public async Task RegisterAsync(WorkerRegistration reg, CancellationToken ct = d { using var cmd = _conn.CreateCommand(); cmd.CommandText = """ - INSERT INTO worker_registry (worker_id, host_name, process_id, started_at, last_heartbeat_at, current_work_item_id, executor_host_id, max_concurrent_sandboxes, executor_network_profiles, executor_credentials, cordoned, healthy) - VALUES ($id, $host, $pid, $started, $hb, $item, $exhost, $cap, $profiles, $creds, $cordoned, $healthy) + INSERT INTO worker_registry (worker_id, host_name, process_id, started_at, last_heartbeat_at, current_work_item_id, executor_host_id, max_concurrent_sandboxes, executor_network_profiles, executor_credentials, executor_capabilities, cordoned, healthy) + VALUES ($id, $host, $pid, $started, $hb, $item, $exhost, $cap, $profiles, $creds, $caps, $cordoned, $healthy) ON CONFLICT(worker_id) DO UPDATE SET host_name = excluded.host_name, process_id = excluded.process_id, @@ -126,6 +127,7 @@ ON CONFLICT(worker_id) DO UPDATE SET max_concurrent_sandboxes = excluded.max_concurrent_sandboxes, executor_network_profiles = excluded.executor_network_profiles, executor_credentials = excluded.executor_credentials, + executor_capabilities = excluded.executor_capabilities, cordoned = excluded.cordoned, healthy = excluded.healthy; """; @@ -400,6 +402,7 @@ private static void Bind(SqliteCommand cmd, WorkerRegistration reg) cmd.Parameters.AddWithValue("$cap", (object?)reg.MaxConcurrentSandboxes ?? DBNull.Value); cmd.Parameters.AddWithValue("$profiles", (object?)SerializeStringList(reg.ExecutorNetworkProfiles) ?? DBNull.Value); cmd.Parameters.AddWithValue("$creds", (object?)SerializeStringList(reg.ExecutorCredentials) ?? DBNull.Value); + cmd.Parameters.AddWithValue("$caps", (object?)SerializeStringList(reg.ExecutorCapabilities) ?? DBNull.Value); cmd.Parameters.AddWithValue("$cordoned", reg.Cordoned ? 1 : 0); cmd.Parameters.AddWithValue("$healthy", reg.Healthy ? 1 : 0); } @@ -416,10 +419,21 @@ private static void Bind(SqliteCommand cmd, WorkerRegistration reg) MaxConcurrentSandboxes = r.IsDBNull(r.GetOrdinal("max_concurrent_sandboxes")) ? null : r.GetInt32(r.GetOrdinal("max_concurrent_sandboxes")), ExecutorNetworkProfiles = r.IsDBNull(r.GetOrdinal("executor_network_profiles")) ? null : DeserializeStringList(r.GetString(r.GetOrdinal("executor_network_profiles"))), ExecutorCredentials = r.IsDBNull(r.GetOrdinal("executor_credentials")) ? null : DeserializeStringList(r.GetString(r.GetOrdinal("executor_credentials"))), + ExecutorCapabilities = HasColumn(r, "executor_capabilities") && !r.IsDBNull(r.GetOrdinal("executor_capabilities")) ? DeserializeStringList(r.GetString(r.GetOrdinal("executor_capabilities"))) : null, Cordoned = r.GetInt32(r.GetOrdinal("cordoned")) != 0, Healthy = r.GetInt32(r.GetOrdinal("healthy")) != 0, }; + private static bool HasColumn(SqliteDataReader r, string column) + { + for (var i = 0; i < r.FieldCount; i++) + { + if (string.Equals(r.GetName(i), column, StringComparison.OrdinalIgnoreCase)) + return true; + } + return false; + } + /// /// Adds the executor-attribute columns to a worker_registry table /// created by an older build. Fresh databases already carry the columns @@ -455,6 +469,7 @@ private static readonly (string Column, string Definition)[] ExecutorColumnDefin ("max_concurrent_sandboxes", "max_concurrent_sandboxes INTEGER"), ("executor_network_profiles", "executor_network_profiles TEXT"), ("executor_credentials", "executor_credentials TEXT"), + ("executor_capabilities", "executor_capabilities TEXT"), ("cordoned", "cordoned INTEGER NOT NULL DEFAULT 0"), ("healthy", "healthy INTEGER NOT NULL DEFAULT 1"), ]; diff --git a/tests/CodeyBox.Tests/ExecutorEligibilityTests.cs b/tests/CodeyBox.Tests/ExecutorEligibilityTests.cs index b79b2f00..04e93d6c 100644 --- a/tests/CodeyBox.Tests/ExecutorEligibilityTests.cs +++ b/tests/CodeyBox.Tests/ExecutorEligibilityTests.cs @@ -13,7 +13,8 @@ private static ExecutorRegistration Host( bool cordoned = false, bool healthy = true, string[]? profiles = null, - string[]? credentials = null) => new() + string[]? credentials = null, + string[]? capabilities = null) => new() { HostId = "exec-1", MaxConcurrentSandboxes = capacity, @@ -21,6 +22,7 @@ private static ExecutorRegistration Host( Healthy = healthy, AllowedNetworkProfiles = profiles ?? [], DeclaredCredentials = credentials ?? [], + DeclaredCapabilities = capabilities ?? [], }; [Fact] @@ -115,4 +117,44 @@ public void WorkerId_IsStableAndPrefixed() Assert.Equal("executor:exec-1", host.WorkerId); Assert.Equal(host.WorkerId, ExecutorRegistration.WorkerIdFor(" exec-1 ")); } + + [Fact] + public void Capabilities_EmptyRequired_CoveredByAnyHost() + { + Assert.True(ExecutorEligibility.CoversRequiredCapabilities(Host(), [])); + Assert.True(ExecutorEligibility.CoversRequiredCapabilities(Host(capabilities: []), [])); + } + + [Fact] + public void Capabilities_CoveredOnlyByExactCaseInsensitiveMatch() + { + var host = Host(capabilities: ["Sensitive"]); + Assert.True(ExecutorEligibility.CoversRequiredCapabilities(host, ["sensitive"])); + Assert.True(ExecutorEligibility.CoversRequiredCapabilities(host, ["SENSITIVE"])); + Assert.False(ExecutorEligibility.CoversRequiredCapabilities(host, ["sensitive-extra"])); + Assert.False(ExecutorEligibility.CoversRequiredCapabilities(host, ["sens"])); + Assert.False(ExecutorEligibility.CoversRequiredCapabilities(Host(capabilities: []), ["sensitive"])); + } + + [Fact] + public void Capabilities_AllRequiredTagsMustBeCovered() + { + var host = Host(capabilities: ["sensitive", "audit"]); + Assert.True(ExecutorEligibility.CoversRequiredCapabilities(host, ["sensitive", "audit"])); + Assert.False(ExecutorEligibility.CoversRequiredCapabilities(host, ["sensitive", "architectural"])); + } + + [Fact] + public void FindCapabilityNoHostProvides_NamesFirstUnmetTag() + { + var hosts = new[] + { + new ExecutorRegistration { HostId = "a", DeclaredCapabilities = ["general"] }, + new ExecutorRegistration { HostId = "b", DeclaredCapabilities = ["sensitive"] }, + }; + Assert.Null(ExecutorEligibility.FindCapabilityNoHostProvides(hosts, [])); + Assert.Null(ExecutorEligibility.FindCapabilityNoHostProvides(hosts, ["sensitive"])); + Assert.Equal("architectural", ExecutorEligibility.FindCapabilityNoHostProvides(hosts, ["sensitive", "architectural"])); + Assert.Equal("nope", ExecutorEligibility.FindCapabilityNoHostProvides(hosts, ["nope"])); + } } diff --git a/tests/CodeyBox.Tests/ExecutorHostTests.cs b/tests/CodeyBox.Tests/ExecutorHostTests.cs index b83b4e02..95d7b9c0 100644 --- a/tests/CodeyBox.Tests/ExecutorHostTests.cs +++ b/tests/CodeyBox.Tests/ExecutorHostTests.cs @@ -38,6 +38,7 @@ public void ExecutorOptions_Validate_AcceptsWellFormed() Assert.Equal(2, registration.MaxConcurrentSandboxes); Assert.Equal(["restricted"], registration.AllowedNetworkProfiles); Assert.Equal(["claude"], registration.DeclaredCredentials); + Assert.Equal(["sensitive"], registration.DeclaredCapabilities); } [Theory] @@ -212,6 +213,7 @@ public async Task Client_Register_SendsBearerAuthAndCapacity() Assert.Equal("exec-1", doc.RootElement.GetProperty("hostId").GetString()); Assert.Equal(2, doc.RootElement.GetProperty("maxConcurrentSandboxes").GetInt32()); Assert.Equal("claude", doc.RootElement.GetProperty("declaredCredentials").EnumerateArray().Single().GetString()); + Assert.Equal("sensitive", doc.RootElement.GetProperty("declaredCapabilities").EnumerateArray().Single().GetString()); } [Fact] @@ -433,6 +435,32 @@ public async Task Server_Register_RejectsInvalidPayloads() declaredCredentials = Enumerable.Range(0, ExecutorRegistration.MaxDeclaredEntries + 1).Select(i => "c" + i).ToArray(), }); Assert.Equal(HttpStatusCode.BadRequest, tooMany.StatusCode); + + var tooManyCapabilities = await api.PostAsJsonAsync("/executors/register", new + { + hostId = "h", + declaredCapabilities = Enumerable.Range(0, ExecutorRegistration.MaxDeclaredEntries + 1).Select(i => "t" + i).ToArray(), + }); + Assert.Equal(HttpStatusCode.BadRequest, tooManyCapabilities.StatusCode); + } + + [Fact] + public async Task Server_Register_PersistsDeclaredCapabilities() + { + using var factory = new ExecutorApiFactory(); + using var api = factory.CreateClient(); + + var resp = await api.PostAsJsonAsync("/executors/register", new + { + hostId = "cap-host", + declaredCapabilities = new[] { "sensitive" }, + }); + resp.EnsureSuccessStatusCode(); + + var workers = await api.GetFromJsonAsync("/workers"); + var row = workers.EnumerateArray() + .Single(w => w.GetProperty("workerId").GetString() == "executor:cap-host"); + Assert.Equal("sensitive", row.GetProperty("executorCapabilities").EnumerateArray().Single().GetString()); } [Fact] @@ -464,6 +492,7 @@ await registry.RegisterAsync(new WorkerRegistration MaxConcurrentSandboxes = 3, ExecutorNetworkProfiles = ["restricted"], ExecutorCredentials = ["claude", "codex"], + ExecutorCapabilities = ["sensitive"], Cordoned = true, Healthy = false, }); @@ -473,6 +502,7 @@ await registry.RegisterAsync(new WorkerRegistration Assert.Equal(3, found.MaxConcurrentSandboxes); Assert.Equal(["restricted"], found.ExecutorNetworkProfiles); Assert.Equal(["claude", "codex"], found.ExecutorCredentials); + Assert.Equal(["sensitive"], found.ExecutorCapabilities); Assert.True(found.Cordoned); Assert.False(found.Healthy); Assert.True(found.IsExecutor); @@ -627,6 +657,7 @@ public async Task Executor_BindsNoInboundPort() MaxConcurrentSandboxes = 2, AllowedNetworkProfiles = ["restricted"], DeclaredCredentials = ["claude"], + DeclaredCapabilities = ["sensitive"], }; private static ExecutorClient MakeClient( diff --git a/tests/CodeyBox.Tests/ExecutorPhaseProxyTests.cs b/tests/CodeyBox.Tests/ExecutorPhaseProxyTests.cs index e566d21a..7b714cf1 100644 --- a/tests/CodeyBox.Tests/ExecutorPhaseProxyTests.cs +++ b/tests/CodeyBox.Tests/ExecutorPhaseProxyTests.cs @@ -293,18 +293,28 @@ public async Task NoExecutorRegistered_FallsBackToInProcess_Unchanged() } [Fact] - public async Task CordonedExecutor_IsNeverSelected_FallsBackToInProcess() + public async Task CordonedExecutor_IsNeverSelected_RequeuedUnderBackoff_NotFailed() { using var ctx = CreateContext(["exec-1"], cordoned: true); var item = WorkItemId.New(); await SeedBareRepoAsync(ctx.Git, item); + var request = NewRequest(item, "work", 0); - var result = await ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0), CancellationToken.None); + var thrown = await Assert.ThrowsAsync( + () => ctx.Proxy.ExecutePhaseAsync(request, CancellationToken.None)); - Assert.Equal(ExecutorPhaseOutcome.Succeeded, result.Outcome); - Assert.NotNull(result.CommitSha); + Assert.Equal("executor", thrown.Provider); + Assert.Equal("placement", thrown.Operation); + Assert.Equal("no-eligible-host", thrown.ErrorClass); + Assert.Contains("exec-1=cordoned", thrown.Detail); + Assert.Equal(TimeSpan.FromSeconds(15), thrown.RecheckIn); Assert.Equal(0, ctx.Factory.Resolves); - Assert.Equal(1, ctx.InnerSpy.Calls); + 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] @@ -313,9 +323,179 @@ 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()); + Assert.Throws(() => new ExecutorPhaseDispatchOptions { PlacementRecheckIn = TimeSpan.Zero }.Validate()); + Assert.Throws(() => new ExecutorPhaseDispatchOptions { RuntimeUnhealthyBackoff = TimeSpan.Zero }.Validate()); new ExecutorPhaseDispatchOptions().Validate(); } + // ── verification 8: credential-aware placement ────────────────────────── + + [Fact] + public async Task CredentialHeldByOneHost_PlacedOnThatHost() + { + using var ctx = CreateContext([]); + AddExecutorHost(ctx, "exec-1", credentials: ["claude"]); + AddExecutorHost(ctx, "exec-2", credentials: ["codex"]); + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + + var result = await ctx.Proxy.ExecutePhaseAsync( + NewPlacementRequest(item, "work", 0, credential: "codex"), CancellationToken.None); + + Assert.Equal(ExecutorPhaseOutcome.Succeeded, result.Outcome); + Assert.Equal(0, ctx.Transports["exec-1"].RunPhaseCalls); + Assert.Equal(1, ctx.Transports["exec-2"].RunPhaseCalls); + } + + // ── verification 9: network-profile placement ─────────────────────────── + + [Fact] + public async Task NetworkProfileAbsentFromHost_NeverPlacedThere() + { + using var ctx = CreateContext([]); + AddExecutorHost(ctx, "exec-1", profiles: ["open"]); + AddExecutorHost(ctx, "exec-2", profiles: ["open", "restricted"]); + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + + var result = await ctx.Proxy.ExecutePhaseAsync( + NewPlacementRequest(item, "work", 0, networkProfile: "restricted"), CancellationToken.None); + + Assert.Equal(ExecutorPhaseOutcome.Succeeded, result.Outcome); + Assert.Equal(0, ctx.Transports["exec-1"].RunPhaseCalls); + Assert.Equal(1, ctx.Transports["exec-2"].RunPhaseCalls); + } + + // ── verification 10: capacity ─────────────────────────────────────────── + + [Fact] + public async Task FullHost_SkippedWhileAtCapacity_SelectableWhenFreed() + { + using var ctx = CreateContext([]); + AddExecutorHost(ctx, "exec-1", capacity: 1, currentWorkItemId: WorkItemId.New().ToString()); + AddExecutorHost(ctx, "exec-2", capacity: 1); + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + + var first = await ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0), CancellationToken.None); + Assert.Equal(ExecutorPhaseOutcome.Succeeded, first.Outcome); + Assert.Equal(0, ctx.Transports["exec-1"].RunPhaseCalls); + Assert.Equal(1, ctx.Transports["exec-2"].RunPhaseCalls); + + await ctx.Registry.HeartbeatAsync(ExecutorRegistration.WorkerIdFor("exec-1"), null); + var freed = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, freed); + var second = await ctx.Proxy.ExecutePhaseAsync(NewRequest(freed, "work", 0), CancellationToken.None); + Assert.Equal(ExecutorPhaseOutcome.Succeeded, second.Outcome); + Assert.Equal(1, ctx.Transports["exec-1"].RunPhaseCalls); + } + + // ── verification 11: unhealthy excluded, requeued under backoff ───────── + + [Fact] + public async Task UnhealthyHost_Excluded_RequeuedUnderBackoff_NotFailed() + { + using var ctx = CreateContext([]); + AddExecutorHost(ctx, "exec-1", healthy: false); + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + var request = NewRequest(item, "work", 0); + + var thrown = await Assert.ThrowsAsync( + () => ctx.Proxy.ExecutePhaseAsync(request, CancellationToken.None)); + + Assert.Equal("executor", thrown.Provider); + Assert.Equal("no-eligible-host", thrown.ErrorClass); + Assert.Contains("exec-1=unhealthy", thrown.Detail); + Assert.Equal(TimeSpan.FromSeconds(15), thrown.RecheckIn); + Assert.Equal(0, ctx.Factory.Resolves); + Assert.Equal(0, ctx.InnerSpy.Calls); + } + + // ── verification 12: unplaceable capability ───────────────────────────── + + [Fact] + public async Task CapabilityNoHostProvides_ReportedUnplaceable_NamingTag_WithoutReworkCost() + { + using var ctx = CreateContext([]); + AddExecutorHost(ctx, "exec-1", capabilities: ["general"]); + AddExecutorHost(ctx, "exec-2", capabilities: ["general"]); + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + var request = NewPlacementRequest(item, "work", 0, capabilities: ["sensitive"]); + + var thrown = await Assert.ThrowsAsync( + () => ctx.Proxy.ExecutePhaseAsync(request, CancellationToken.None)); + + Assert.Equal("sensitive", thrown.UnmetCapability); + Assert.Contains("exec-1=missing-capability:sensitive", thrown.Detail); + Assert.Contains("exec-2=missing-capability:sensitive", thrown.Detail); + Assert.NotNull(thrown.Decision); + Assert.True(thrown.Decision.IsUnplaceable); + Assert.Equal(0, ctx.Factory.Resolves); + 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); + } + + // ── verification 13: refusal reasons recorded per candidate ───────────── + + [Fact] + public async Task NoEligibleHost_DeferralNames_EachCandidateRefusalReason() + { + using var ctx = CreateContext([]); + AddExecutorHost(ctx, "exec-1", cordoned: true); + AddExecutorHost(ctx, "exec-2", capacity: 1, currentWorkItemId: WorkItemId.New().ToString()); + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + + var thrown = await Assert.ThrowsAsync( + () => ctx.Proxy.ExecutePhaseAsync(NewRequest(item, "work", 0), CancellationToken.None)); + + Assert.Equal("no-eligible-host", thrown.ErrorClass); + Assert.Contains("exec-1=cordoned", thrown.Detail); + Assert.Contains("exec-2=at-capacity(1/1)", thrown.Detail); + } + + // ── verification 14: lying host fails over, failure host-attributed ───── + + [Fact] + public async Task HostDeclaringCredentialItLacks_FailsOverToEligibleHost_NotAgentFailure() + { + using var ctx = CreateContext([]); + AddExecutorHost(ctx, "exec-1", credentials: ["codex"]); + AddExecutorHost(ctx, "exec-2", credentials: ["codex"]); + ctx.Transports["exec-1"].RunPhaseOverride = _ => new ExecutorPhaseResult + { + Outcome = ExecutorPhaseOutcome.AgentFailed, + Usage = new ExecutorPhaseUsage(1, 1, 0m), + Findings = [], + ErrorMessage = ExecutorPhaseProxy.CredentialMissingErrorPrefix + " credential file absent on host", + }; + var item = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, item); + + var result = await ctx.Proxy.ExecutePhaseAsync( + NewPlacementRequest(item, "work", 0, credential: "codex"), CancellationToken.None); + + Assert.Equal(ExecutorPhaseOutcome.Succeeded, result.Outcome); + Assert.NotNull(result.CommitSha); + Assert.Equal(1, ctx.Transports["exec-1"].RunPhaseCalls); + Assert.Equal(1, ctx.Transports["exec-2"].RunPhaseCalls); + Assert.Equal(0, ctx.InnerSpy.Calls); + + var next = WorkItemId.New(); + await SeedBareRepoAsync(ctx.Git, next); + var second = await ctx.Proxy.ExecutePhaseAsync( + NewPlacementRequest(next, "work", 0, credential: "codex"), CancellationToken.None); + Assert.Equal(ExecutorPhaseOutcome.Succeeded, second.Outcome); + Assert.Equal(1, ctx.Transports["exec-1"].RunPhaseCalls); + Assert.Equal(2, ctx.Transports["exec-2"].RunPhaseCalls); + } + // ── harness ───────────────────────────────────────────────────────────── private static ExecutorPhaseRequest NewRequest(WorkItemId item, string phase, int attempt, string payload = "{}") => @@ -347,14 +527,46 @@ private TestHarness CreateContext(string[] executors, long? maxArchiveBytes = nu var inner = new InProcessExecutorPhaseRunner(git, handler, () => options); var spy = new SpyRunner(inner); var proxy = new ExecutorPhaseProxy(registry, factory, git, store, spy, () => options); + var ctx = new TestHarness(git, store, registry, handler, factory, inner, spy, proxy); 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); + AddExecutorHost(ctx, host, cordoned: cordoned); + return ctx; } + private void AddExecutorHost( + TestHarness ctx, + string hostId, + bool cordoned = false, + bool healthy = true, + int? capacity = 4, + string[]? profiles = null, + string[]? credentials = null, + string[]? capabilities = null, + string? currentWorkItemId = null) + { + ctx.Registry.AddExecutor(hostId, cordoned, healthy, capacity, profiles, credentials, capabilities, currentWorkItemId); + ctx.Factory.AddHost(hostId, Path.Combine(_root, "executor-" + hostId + "-" + Guid.NewGuid().ToString("N")), ctx.Handler); + } + + private static ExecutorPhaseRequest NewPlacementRequest( + WorkItemId item, + string phase, + int attempt, + string? credential = null, + string? networkProfile = null, + string[]? capabilities = null, + string payload = "{}") => new() + { + WorkItemId = item.ToString(), + Phase = phase, + Attempt = attempt, + RepositoryId = item.ToString(), + PayloadJson = payload, + RequiredCredential = credential, + RequiredNetworkProfile = networkProfile, + RequiredCapabilities = capabilities ?? [], + }; + private async Task SeedBareRepoAsync(LocalGitHost git, WorkItemId item) { var repoId = await git.EnsureRepositoryAsync(item, seedFromUrl: null); @@ -454,6 +666,7 @@ private sealed class TestHarness : IDisposable public TestHarness( LocalGitHost git, SqliteIdempotencyStore store, + FakeWorkerRegistry registry, GitCommitPhaseHandler handler, FakeTransportFactory factory, InProcessExecutorPhaseRunner inner, @@ -462,6 +675,7 @@ public TestHarness( { Git = git; Store = store; + Registry = registry; Handler = handler; Factory = factory; Inner = inner; @@ -471,6 +685,7 @@ public TestHarness( public LocalGitHost Git { get; } public SqliteIdempotencyStore Store { get; } + public FakeWorkerRegistry Registry { get; } public GitCommitPhaseHandler Handler { get; } public FakeTransportFactory Factory { get; } public Dictionary Transports => Factory.Transports; @@ -500,7 +715,15 @@ private sealed class FakeWorkerRegistry : IWorkerRegistry { private readonly Dictionary _rows = new(StringComparer.Ordinal); - public void AddExecutor(string hostId, bool cordoned = false) + public void AddExecutor( + string hostId, + bool cordoned = false, + bool healthy = true, + int? capacity = 4, + string[]? profiles = null, + string[]? credentials = null, + string[]? capabilities = null, + string? currentWorkItemId = null) { var now = DateTimeOffset.UtcNow; _rows[ExecutorRegistration.WorkerIdFor(hostId)] = new WorkerRegistration @@ -510,12 +733,14 @@ public void AddExecutor(string hostId, bool cordoned = false) ProcessId = 4242, StartedAt = now, LastHeartbeatAt = now, + CurrentWorkItemId = currentWorkItemId, ExecutorHostId = hostId, - MaxConcurrentSandboxes = 4, - ExecutorNetworkProfiles = [], - ExecutorCredentials = [], + MaxConcurrentSandboxes = capacity, + ExecutorNetworkProfiles = profiles ?? [], + ExecutorCredentials = credentials ?? [], + ExecutorCapabilities = capabilities ?? [], Cordoned = cordoned, - Healthy = true, + Healthy = healthy, }; } @@ -650,6 +875,7 @@ public FakePhaseTransport(string hostId, string executorRoot, GitCommitPhaseHand public int RunPhaseCalls { get; private set; } public ExecutorPhaseTransportException? FailWith; public Func? CustomArchive; + public Func? RunPhaseOverride; public long? LastStageOutMaxBytes { get; private set; } public long StageOutBytesWritten { get; private set; } @@ -666,6 +892,15 @@ public Task StageInAsync(string hostRepoPath, CancellationToken ct) { ThrowIfFailing(); StagedInPaths.Add(Path.GetFullPath(hostRepoPath)); + foreach (var entry in Directory.GetFileSystemEntries(ExecutorRoot)) + { + try + { + if (Directory.Exists(entry)) Directory.Delete(entry, recursive: true); + else File.Delete(entry); + } + catch { } + } var dest = Path.Combine(ExecutorRoot, Path.GetFileName(hostRepoPath.TrimEnd(Path.DirectorySeparatorChar))); CopyDirectory(hostRepoPath, dest); return Task.CompletedTask; @@ -674,6 +909,12 @@ public Task StageInAsync(string hostRepoPath, CancellationToken ct) public Task RunPhaseAsync(ExecutorPhaseRequest request, CancellationToken ct) { ThrowIfFailing(); + var overridden = RunPhaseOverride?.Invoke(request); + if (overridden is not null) + { + RunPhaseCalls++; + return Task.FromResult(overridden); + } RunPhaseCalls++; var staged = StagedCopy ?? throw new InvalidOperationException("No staged repo on fake executor."); return _handler.ExecuteAsync(request, staged, ct); diff --git a/tests/CodeyBox.Tests/ExecutorPlacementTests.cs b/tests/CodeyBox.Tests/ExecutorPlacementTests.cs new file mode 100644 index 00000000..fe70cd3e --- /dev/null +++ b/tests/CodeyBox.Tests/ExecutorPlacementTests.cs @@ -0,0 +1,204 @@ +using CodeyBox.Core; + +namespace CodeyBox.Tests; + +/// +/// Pure placement-decider verification: credential, network profile, +/// capability, capacity, cordon/health and observability rules compose in +/// without any I/O. +/// +public sealed class ExecutorPlacementTests +{ + private static ExecutorRegistration Host( + string id = "exec-1", + int? capacity = 4, + bool cordoned = false, + bool healthy = true, + string[]? profiles = null, + string[]? credentials = null, + string[]? capabilities = null) => new() + { + HostId = id, + MaxConcurrentSandboxes = capacity, + Cordoned = cordoned, + Healthy = healthy, + AllowedNetworkProfiles = profiles ?? [], + DeclaredCredentials = credentials ?? [], + DeclaredCapabilities = capabilities ?? [], + }; + + private static ExecutorPlacementRequirements NoRequirements() => new() + { + RequiredCredential = null, + RequiredNetworkProfile = null, + RequiredCapabilities = [], + }; + + [Fact] + public void CredentialHeldByOneHost_SelectsThatHost() + { + var hosts = new[] + { + Host("exec-1", credentials: ["claude"]), + Host("exec-2", credentials: ["codex"]), + }; + var decision = ExecutorPlacement.Decide(hosts, NoRequirements() with { RequiredCredential = "codex" }); + + Assert.Equal("exec-2", decision.SelectedHostId); + Assert.False(decision.IsUnplaceable); + Assert.Contains(decision.Candidates, c => c.HostId == "exec-1" && c.Reason == "missing-credential:codex"); + Assert.Contains(decision.Candidates, c => c.HostId == "exec-2" && c.Reason == "selected"); + } + + [Fact] + public void Credential_MatchesByExactEqualityOnly() + { + var hosts = new[] { Host(credentials: ["codex"]) }; + var decision = ExecutorPlacement.Decide(hosts, NoRequirements() with { RequiredCredential = "codex-admin" }); + + Assert.Null(decision.SelectedHostId); + Assert.Contains(decision.Candidates, c => c.Reason == "missing-credential:codex-admin"); + } + + [Fact] + public void NetworkProfileAbsentFromHost_NeverSelected() + { + var hosts = new[] + { + Host("exec-1", profiles: ["open"]), + Host("exec-2", profiles: ["open", "restricted"]), + }; + var decision = ExecutorPlacement.Decide(hosts, NoRequirements() with { RequiredNetworkProfile = "restricted" }); + + Assert.Equal("exec-2", decision.SelectedHostId); + Assert.Contains(decision.Candidates, c => c.HostId == "exec-1" && c.Reason == "network-profile:restricted"); + } + + [Fact] + public void HostAtCapacity_NotSelected() + { + var hosts = new[] + { + Host("exec-1", capacity: 1), + Host("exec-2", capacity: 1), + }; + var loads = new Dictionary(StringComparer.Ordinal) { ["exec-1"] = 1 }; + + var full = ExecutorPlacement.Decide(hosts, NoRequirements(), loads); + Assert.Equal("exec-2", full.SelectedHostId); + Assert.Contains(full.Candidates, c => c.HostId == "exec-1" && c.Reason == "at-capacity(1/1)"); + + var freed = ExecutorPlacement.Decide(hosts, NoRequirements(), new Dictionary(StringComparer.Ordinal)); + Assert.Equal("exec-1", freed.SelectedHostId); + } + + [Fact] + public void CordonedAndUnhealthyHosts_Excluded() + { + var hosts = new[] + { + Host("exec-1", cordoned: true), + Host("exec-2", healthy: false), + Host("exec-3"), + }; + var decision = ExecutorPlacement.Decide(hosts, NoRequirements()); + + Assert.Equal("exec-3", decision.SelectedHostId); + Assert.Contains(decision.Candidates, c => c.HostId == "exec-1" && c.Reason == "cordoned"); + Assert.Contains(decision.Candidates, c => c.HostId == "exec-2" && c.Reason == "unhealthy"); + } + + [Fact] + public void NoEligibleHost_TransientWhenCapabilityHeldSomewhere() + { + var hosts = new[] + { + Host("exec-1", cordoned: true, capabilities: ["sensitive"]), + }; + var requirements = NoRequirements() with { RequiredCapabilities = (IReadOnlyList)["sensitive"] }; + var decision = ExecutorPlacement.Decide(hosts, requirements); + + Assert.Null(decision.SelectedHostId); + Assert.False(decision.IsUnplaceable); + Assert.Null(decision.UnmetCapability); + } + + [Fact] + public void CapabilityNoHostProvides_IsUnplaceableNamingTag() + { + var hosts = new[] + { + Host("exec-1", capabilities: ["general"]), + Host("exec-2", capabilities: ["general"]), + }; + var requirements = NoRequirements() with { RequiredCapabilities = (IReadOnlyList)["sensitive"] }; + var decision = ExecutorPlacement.Decide(hosts, requirements); + + Assert.Null(decision.SelectedHostId); + Assert.True(decision.IsUnplaceable); + Assert.Equal("sensitive", decision.UnmetCapability); + Assert.All(decision.Candidates, c => Assert.Equal($"missing-capability:sensitive", c.Reason)); + } + + [Fact] + public void Capabilities_MatchCaseInsensitively_InExistingVocabulary() + { + var hosts = new[] { Host(capabilities: ["Sensitive"]) }; + var requirements = NoRequirements() with { RequiredCapabilities = (IReadOnlyList)["sensitive"] }; + var decision = ExecutorPlacement.Decide(hosts, requirements); + + Assert.Equal("exec-1", decision.SelectedHostId); + } + + [Fact] + public void LeastLoadedHost_Wins_TiesBreakByHostId() + { + var hosts = new[] + { + Host("exec-b", capacity: 4), + Host("exec-a", capacity: 4), + }; + var loads = new Dictionary(StringComparer.Ordinal) { ["exec-a"] = 1, ["exec-b"] = 3 }; + + var byLoad = ExecutorPlacement.Decide(hosts, NoRequirements(), loads); + Assert.Equal("exec-a", byLoad.SelectedHostId); + + var tie = ExecutorPlacement.Decide(hosts, NoRequirements()); + Assert.Equal("exec-a", tie.SelectedHostId); + } + + [Fact] + public void RuntimeUnhealthyHosts_Skipped() + { + var hosts = new[] { Host("exec-1"), Host("exec-2") }; + var backedOff = new HashSet(StringComparer.Ordinal) { "exec-1" }; + var decision = ExecutorPlacement.Decide(hosts, NoRequirements(), runtimeUnhealthy: backedOff); + + Assert.Equal("exec-2", decision.SelectedHostId); + Assert.Contains(decision.Candidates, c => c.HostId == "exec-1" && c.Reason == "runtime-unhealthy"); + } + + [Fact] + public void Decision_Describe_NamesSelectionAndRefusals() + { + var hosts = new[] + { + Host("exec-1", cordoned: true), + Host("exec-2"), + }; + var decision = ExecutorPlacement.Decide(hosts, NoRequirements()); + + Assert.Contains("selected=exec-2", decision.Describe()); + Assert.Contains("exec-1=cordoned", decision.Describe()); + } + + [Fact] + public void EmptyHosts_SelectsNothing_WithoutUnplaceable() + { + var decision = ExecutorPlacement.Decide([], NoRequirements()); + + Assert.Null(decision.SelectedHostId); + Assert.False(decision.IsUnplaceable); + Assert.Empty(decision.Candidates); + } +}