diff --git a/docs/operating/quota.md b/docs/operating/quota.md index 8932cab0..0b99c26d 100644 --- a/docs/operating/quota.md +++ b/docs/operating/quota.md @@ -127,6 +127,54 @@ refused (never dispatched on an unkeyed reading) and the reason names the member and the unresolved pool. `/quota` reports each member's `pool` and `poolKind` so operators can see which members share a meter. +## Executor-reported pools: metering a credential the orchestrator cannot read + +By default every pool is orchestrator-probed: the orchestrator holds the +credential and probes directly in-process. When the credential for an +account lives on an executor host the orchestrator cannot read, that +account has no reading — so the pool can instead be metered from readings +the executor reports: + +```json +"QuotaRouter": { + "Pools": { + "edge-sub": { + "Kind": "ResettingWindow", + "ProbeSource": "ExecutorReported", + "ReportedReadingMaxAgeSeconds": 300, + "HolderHostIds": ["edge-gpu-01"] + } + } +} +``` + +The holding executor probes locally with its own credential and POSTs each +reading (pool, available percentage or balance, reset time, observed time) +to `/executors/{hostId}/quota-reports`. The request must carry that host's +own token: each executor gets a named `CodeyBox:ApiClients` entry whose +`ExecutorHostId` equals its host id, and the executor uses that token (not +the shared operator key) as its API bearer. The shared operator key and any +token without a host binding are rejected on this endpoint with `403`, and +a bound token reporting for a different host is rejected too — the path +host is matched against the authenticated caller's binding before the +registry is even consulted, so a rejected caller cannot probe which host +ids are registered. The orchestrator validates every +report on arrival — the pool must exist and be executor-reported, the +reporting host must be declared in the pool's `HolderHostIds` (exact match; +a report from any other host is rejected and the stored reading is left +unchanged), percentages must sit within 0–100, balances must be finite and +non-negative, and a depleting-balance pool never carries a reset instant — +and meters the pool from the latest fresh report. + +The orchestrator stays the sole authority for the gate decision: an +executor reports readings and never decides admission. A stale report +(older than `ReportedReadingMaxAgeSeconds`) or a pool that never reported +reads as unknown — never as healthy headroom — and flows through the same +unknown handling as a direct probe, including the Transient / Permanent / +NoCredential distinction and fail-closed whenever a non-zero floor is in +force. `ReportedReadingClockSkewSeconds` (default 300) bounds how far in +the future a report's observed time may be before it is rejected. + ## Replenishment kinds: resetting windows vs depleting balances Two kinds of allowance are in use and they are not interchangeable: diff --git a/docs/reference/api.md b/docs/reference/api.md index dacc5ed8..d2540169 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -31,6 +31,16 @@ at least 32 characters or the service refuses to start. "Subject": "service", "DisplayName": "JobTrack" } + }, + { + "Name": "exec-1", + "TokenEnvVar": "CODEYBOX_EXEC_1_API_KEY", + "ExecutorHostId": "exec-1", + "Principal": { + "Issuer": "codeybox", + "Subject": "exec-1", + "DisplayName": "Executor exec-1" + } } ] } @@ -44,6 +54,14 @@ Delegated identities are authenticated claims from the integration—not credentials—and are persisted with the work item for API, commit, and pull request attribution. +A client with `ExecutorHostId` is a host-bound executor token: it may act +only as that executor host on host-scoped executor endpoints (notably +`POST /executors/{hostId}/quota-reports`, which rejects any other caller — +including the operator key — so a shared bearer cannot forge another host's +quota meter). Give each executor host its own token environment variable and +matching `ExecutorHostId`, and put that token (not the operator key) in the +executor's `ApiKeyEnvVar` on that host. + ### GitHub App delivery credentials For team installations, configure the GitHub upstream with a GitHub App @@ -1398,6 +1416,21 @@ Heartbeat a registered executor into the worker registry. Request body is `{ "cu Remove an executor registration (clean shutdown). Response: `200 OK` with `{ "hostId": "exec-1" }`. +### `POST /executors/{hostId}/quota-reports` + +Report one quota reading for a pool whose credential the calling host holds +(pool identity, availability reading, reset time, observed time). The bearer +must be a per-executor token bound to the path host (see `ExecutorHostId` +under [Authentication](#authentication)); the shared operator key and any +token without a host binding are rejected, so one bearer holder cannot forge +another host's meter. The host must also be registered and be declared in +the pool's `HolderHostIds`. Response: `200 OK` with `{ "accepted": true, +"pool": "" }`. `401` with no bearer, `403` on a token/host mismatch, +`404` for an unregistered host, `400` on a rejected report (unknown pool, +non-holder, out-of-range reading, inconsistent reset). Accepting a report +never decides admission — the orchestrator's quota gate does that. See +[`quota.md`](../operating/quota.md) for the executor-reported pool design. + ### `GET /sandboxes/leaked` Returns the provider-owned persistent sandboxes detected as leaked on the most diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index 7b164334..d12d3863 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -718,7 +718,7 @@ Tuning knobs for the quota probe and deferred-requeue logic. | `EndFloorPct` | `3` | Global late-window ramp floor as reset approaches. | | `RampWindowSeconds` | `604800` | Global quota-window length used for the ramp calculation. | | `FloorByAgent` | `{}` | Optional per-agent overrides keyed by agent kind, e.g. `codex` or `claude`. Each entry may set `StartFloorPct`, `EndFloorPct`, `MinQuotaPct`, and `RampWindowSeconds`; omitted fields inherit global values, and omitted agents use the global ramp. | -| `Pools` | `{}` | Optional quota pools keyed by pool name. Each pool names one underlying account or subscription; class members join via their `Pool` reference and share one reading, one floor, and one reservation escrow. Each entry sets `Kind` (`ResettingWindow` or `DepletingBalance`), optional `BalanceUnit`, and optional `ReservationEstimate` (native units). Hot-reloadable. See `docs/operating/quota.md`. | +| `Pools` | `{}` | Optional quota pools keyed by pool name. Each pool names one underlying account or subscription; class members join via their `Pool` reference and share one reading, one floor, and one reservation escrow. Each entry sets `Kind` (`ResettingWindow` or `DepletingBalance`), optional `BalanceUnit`, and optional `ReservationEstimate` (native units). Each entry may also set `ProbeSource` (`OrchestratorDirect`, the default, or `ExecutorReported` for accounts whose credential lives on an executor host), `ReportedReadingMaxAgeSeconds` (staleness bound for executor reports, default `300`), and `HolderHostIds` (executor hosts authorised to report for the pool; reports from other hosts are rejected). Hot-reloadable. See `docs/operating/quota.md`. | | `FloorByPool` | `{}` | Optional per-pool floor overrides keyed by pool name, alongside `FloorByAgent`. For a pool member the higher of the pool-resolved and agent-resolved floors wins. Resetting-window pools use the percentage fields (`MinQuotaPct`, `StartFloorPct`, `EndFloorPct`, `RampWindowSeconds`); depleting-balance pools use absolute `MinBalance`. Mixing units is rejected at load. Hot-reloadable. | | `QuotaRecheckIntervalSeconds` | `300` | Seconds to wait before re-probing when all Subscription members are exhausted. | | `QuotaCacheTtlSeconds` | `60` | Seconds to cache a quota probe result (per probe instance). | @@ -730,6 +730,7 @@ Tuning knobs for the quota probe and deferred-requeue logic. | `ExpectedResets` | `{}` | Optional per-agent expected reset declarations, keyed by agent kind. Each entry may set explicit `Timestamps` and/or a recurring `CadenceSeconds` with `CadenceAnchor`; the policy paces to the sooner of the live probe reset and the next expected reset. Hot-reloadable. | | `ObservedFailureWindowMinutes` | `10` | Minutes a recent quota-shaped failure blocks the same agent/model across all projects. | | `ObservedFailureRetentionMinutes` | `30` | Minutes observed quota failures remain in `state.db`. | +| `ReportedReadingClockSkewSeconds` | `300` | Tolerance in seconds for clock skew between executor and orchestrator hosts when validating an executor-reported reading's observed time. A report dated further in the future than this is rejected. Hot-reloadable. | | `ProbeMaxRetries` | `2` | Additional retries on a transient probe failure (network error / timeout / 5xx) before recording the failure. Hot-reloadable; currently honoured by the Claude probe. | | `ProbeRetryInitialDelayMs` | `250` | Base retry backoff in milliseconds; doubles each attempt. Hot-reloadable. | | `ProbeRetryMaxDelaySeconds` | `300` | Maximum between-retry delay, including provider `Retry-After` values on Anthropic OAuth usage requests. Hot-reloadable. | diff --git a/src/CodeyBox.Api/AgentConfigHotReload.cs b/src/CodeyBox.Api/AgentConfigHotReload.cs index c717ec6f..cf79c0ab 100644 --- a/src/CodeyBox.Api/AgentConfigHotReload.cs +++ b/src/CodeyBox.Api/AgentConfigHotReload.cs @@ -1267,6 +1267,11 @@ private static string SerializeQuotaRouter(QuotaRouterConfig opts) Kind = kv.Value.Kind.ToString(), kv.Value.BalanceUnit, kv.Value.ReservationEstimate, + ProbeSource = kv.Value.ProbeSource.ToString(), + ReportedReadingMaxAgeSeconds = checked((int)kv.Value.ReportedReadingMaxAge.TotalSeconds), + HolderHostIds = kv.Value.HolderHostIds + .OrderBy(h => h, StringComparer.Ordinal) + .ToArray(), }), FloorByPool = mapped.FloorByPool .OrderBy(kv => kv.Key, StringComparer.OrdinalIgnoreCase) @@ -1289,6 +1294,7 @@ private static string SerializeQuotaRouter(QuotaRouterConfig opts) UnknownPolicy = opts.UnknownPolicy.ToString(), opts.ObservedFailureWindowMinutes, opts.ObservedFailureRetentionMinutes, + opts.ReportedReadingClockSkewSeconds, opts.CapRetryIntervalSeconds, opts.ColdStartFitInWindow, opts.DrainAggressiveness, diff --git a/src/CodeyBox.Api/ApiKeyAuth.cs b/src/CodeyBox.Api/ApiKeyAuth.cs index 06f64890..3140321e 100644 --- a/src/CodeyBox.Api/ApiKeyAuth.cs +++ b/src/CodeyBox.Api/ApiKeyAuth.cs @@ -59,8 +59,9 @@ public static void Configure(WebApplicationBuilder builder) throw new InvalidOperationException( $"{client.TokenEnvVar} must contain at least 32 characters of high-entropy random data."); ValidateInitiator(client.Principal); + var executorHostId = NormalizeExecutorHostId(client); resolved.Add(new ResolvedApiClient( - client.Name, token, client.Principal, client.CanDelegateInitiator)); + client.Name, token, client.Principal, client.CanDelegateInitiator, executorHostId)); } return new ApiKeyState(Token: key, Disabled: false, Clients: resolved); @@ -82,7 +83,7 @@ public static IApplicationBuilder UseApiKeyAuth( if (state.Disabled) { ctx.Items[PrincipalItemKey] = new ApiClientPrincipal( - "authentication-disabled", OperatorInitiator, CanDelegateInitiator: false); + AuthenticationDisabledClientName, OperatorInitiator, CanDelegateInitiator: false); await next(); return; } @@ -133,6 +134,25 @@ public static bool IsAuthorized(HttpContext ctx, ApiKeyState state) return state.Disabled || TryAuthenticate(ctx, state, out _); } + internal static bool TryGetPrincipal(HttpContext context, out ApiClientPrincipal? principal) + { + ArgumentNullException.ThrowIfNull(context); + if (context.Items.TryGetValue(PrincipalItemKey, out var value) + && value is ApiClientPrincipal typed) + { + principal = typed; + return true; + } + principal = null; + return false; + } + + internal static bool IsAuthenticationDisabled(ApiClientPrincipal principal) + { + ArgumentNullException.ThrowIfNull(principal); + return string.Equals(principal.Name, AuthenticationDisabledClientName, StringComparison.Ordinal); + } + internal static InitiatorResolution ResolveInitiator( HttpContext context, WorkInitiator? delegated) @@ -175,12 +195,27 @@ private static bool TryAuthenticate( if (!ConstantTimeEquals(presented, client.Token)) continue; principal = new ApiClientPrincipal( - client.Name, client.FixedInitiator, client.CanDelegateInitiator); + client.Name, client.FixedInitiator, client.CanDelegateInitiator, client.ExecutorHostId); return true; } return false; } + private static string? NormalizeExecutorHostId(ApiClientOptions client) + { + if (string.IsNullOrWhiteSpace(client.ExecutorHostId)) + return null; + var trimmed = client.ExecutorHostId.Trim(); + if (trimmed.Length > ExecutorRegistration.MaxHostIdLength) + throw new InvalidOperationException( + $"CodeyBox:ApiClients entry '{client.Name}': ExecutorHostId must be at most " + + $"{ExecutorRegistration.MaxHostIdLength} characters."); + if (trimmed.Any(char.IsControl)) + throw new InvalidOperationException( + $"CodeyBox:ApiClients entry '{client.Name}': ExecutorHostId must not contain control characters."); + return trimmed; + } + private static void ValidateInitiator(WorkInitiator initiator) { ValidateIdentityPart(initiator.Issuer, nameof(initiator.Issuer), 200); @@ -205,6 +240,14 @@ private static void ValidateIdentityPart(string value, string name, int maximumL } internal const string PrincipalItemKey = "CodeyBox.ApiClientPrincipal"; + + /// + /// Client name assigned to requests served while authentication is + /// disabled (CodeyBox:DangerouslyDisableAuth=true, loopback dev + /// only). Such callers carry no token and therefore no executor binding; + /// host-scoped endpoints treat them as the local operator. + /// + internal const string AuthenticationDisabledClientName = "authentication-disabled"; private static readonly WorkInitiator OperatorInitiator = new() { Issuer = "codeybox", @@ -240,12 +283,14 @@ internal sealed record ResolvedApiClient( string Name, string Token, WorkInitiator FixedInitiator, - bool CanDelegateInitiator); + bool CanDelegateInitiator, + string? ExecutorHostId); internal sealed record ApiClientPrincipal( string Name, WorkInitiator FixedInitiator, - bool CanDelegateInitiator); + bool CanDelegateInitiator, + string? ExecutorHostId = null); internal sealed record InitiatorResolution(WorkInitiator? Value, IResult? Error); @@ -255,6 +300,17 @@ public sealed class ApiClientOptions public string TokenEnvVar { get; set; } = string.Empty; public WorkInitiator? Principal { get; set; } public bool CanDelegateInitiator { get; set; } + + /// + /// Optional executor host this client's token is bound to. When set, the + /// token may only act as that host on host-scoped executor endpoints + /// (notably POST /executors/{hostId}/quota-reports): a path host + /// that does not exactly equal this value is rejected, so one executor + /// cannot forge another host's quota meter. Tokens without a binding + /// (including the operator key) are rejected on quota-report ingress — + /// a shared bearer proves nothing about which host is calling. + /// + public string? ExecutorHostId { get; set; } } /// diff --git a/src/CodeyBox.Api/ExecutorEndpoints.cs b/src/CodeyBox.Api/ExecutorEndpoints.cs index e190ea5f..4be0edea 100644 --- a/src/CodeyBox.Api/ExecutorEndpoints.cs +++ b/src/CodeyBox.Api/ExecutorEndpoints.cs @@ -1,4 +1,5 @@ using CodeyBox.Core; +using CodeyBox.Orchestrator; namespace CodeyBox.Api; @@ -26,6 +27,7 @@ public static void Map(WebApplication app) app.MapPost("/executors/register", RegisterAsync); app.MapPost("/executors/{hostId}/heartbeat", HeartbeatAsync); app.MapPost("/executors/{hostId}/deregister", DeregisterAsync); + app.MapPost("/executors/{hostId}/quota-reports", ReportQuotaAsync); } private static async Task RegisterAsync( @@ -155,6 +157,107 @@ internal static string NormalizeHostId(string? hostId) return trimmed; } + /// + /// Ingests one executor-reported quota reading for a pool whose credential + /// the reporting host holds. The caller is bound to the claimed host in + /// three layers: the bearer must be a per-executor token bound to the + /// path host (a shared bearer such as the operator key proves nothing + /// about which host is calling, so it is rejected here — see + /// ApiClientOptions.ExecutorHostId); the host must have a live + /// worker-registry registration (checked here at ingress — an unregistered + /// host has no 404-free path to this store); and it must be + /// operator-declared in the pool's HolderHostIds (checked by the + /// store, which rejects anything else without mutating the stored + /// reading). The caller check runs before the registry lookup so a + /// rejected caller cannot probe which host ids are registered. The store + /// further validates values and reset consistency. This endpoint accepts + /// or rejects reports only — admission is decided by the orchestrator's + /// quota gate, never here. + /// + private static async Task ReportQuotaAsync( + string hostId, + ExecutorQuotaReportRequest? req, + ExecutorQuotaReportStore store, + IWorkerRegistry registry, + HttpContext httpContext, + CancellationToken ct) + { + string normalized; + try + { + normalized = NormalizeHostId(hostId); + } + catch (ArgumentException ex) + { + return Results.BadRequest(new { error = ex.Message }); + } + + if (CheckQuotaReportCaller(httpContext, normalized) is { } callerRejection) + return callerRejection; + + if (req is null) + return Results.BadRequest(new { error = "request body is required" }); + + var workerId = ExecutorRegistration.WorkerIdFor(normalized); + var workers = await registry.ListAsync(ct); + if (!workers.Any(w => string.Equals(w.WorkerId, workerId, StringComparison.Ordinal))) + return Results.NotFound(new { error = $"no executor registered for host id '{normalized}'" }); + + QuotaUnknownReason? unknown = null; + if (!string.IsNullOrWhiteSpace(req.Unknown)) + { + if (!Enum.TryParse(req.Unknown.Trim(), ignoreCase: true, out var parsed) + || !Enum.IsDefined(parsed)) + return Results.BadRequest( + new { error = $"unknown must be one of {string.Join(", ", Enum.GetNames())}" }); + unknown = parsed; + } + + var report = new ExecutorQuotaReport + { + PoolName = req.Pool ?? string.Empty, + AvailablePct = req.AvailablePct, + BalanceRemaining = req.BalanceRemaining, + ResetAt = req.ResetAt, + ObservedAt = req.ObservedAt ?? default, + Unknown = unknown, + Notes = req.Notes, + }; + + if (!store.TryReport(normalized, report, out var rejectionReason)) + return Results.BadRequest(new { error = rejectionReason }); + + var pool = QuotaPoolResolver.NormalizePoolName(report.PoolName) ?? string.Empty; + if (store.TryGetStored(pool, out var stored, out _) && stored?.PoolName is { } storedName) + pool = storedName; + return Results.Ok(new { accepted = true, pool }); + } + + /// + /// Binds the quota-report path host to the authenticated caller. A + /// host-bound executor token may report only for its own host; anything + /// else — a missing principal, a token with no host binding (including + /// the shared operator key), or a bound token calling for a different + /// host — is rejected. Returns null when the caller may proceed. + /// Pure apart from reading the already-authenticated principal. + /// + internal static IResult? CheckQuotaReportCaller(HttpContext httpContext, string normalizedHostId) + { + if (!ApiKeyAuth.TryGetPrincipal(httpContext, out var principal) || principal is null) + return Results.Unauthorized(); + if (ApiKeyAuth.IsAuthenticationDisabled(principal)) + return null; + if (string.IsNullOrWhiteSpace(principal.ExecutorHostId)) + return Results.Json( + new { error = "quota reports require a host-bound executor token (CodeyBox:ApiClients ExecutorHostId); shared bearer tokens cannot report readings" }, + statusCode: StatusCodes.Status403Forbidden); + if (!string.Equals(principal.ExecutorHostId, normalizedHostId, StringComparison.Ordinal)) + return Results.Json( + new { error = $"this token is bound to executor host '{principal.ExecutorHostId}' and cannot report for host '{normalizedHostId}'" }, + statusCode: StatusCodes.Status403Forbidden); + return null; + } + internal static string? ValidateCapacity(int? capacity) { if (capacity is null) @@ -204,4 +307,21 @@ public sealed class ExecutorHeartbeatRequest { public string? CurrentWorkItemId { get; set; } } + + /// + /// One executor-reported quota reading. The pool identity, the availability + /// reading, the reset time, and the time it was observed travel here; the + /// orchestrator validates and stores the report and keeps the gate + /// decision for itself. + /// + public sealed class ExecutorQuotaReportRequest + { + public string? Pool { get; set; } + public double? AvailablePct { get; set; } + public double? BalanceRemaining { get; set; } + public DateTimeOffset? ResetAt { get; set; } + public DateTimeOffset? ObservedAt { get; set; } + public string? Unknown { get; set; } + public string? Notes { get; set; } + } } diff --git a/src/CodeyBox.Api/Program.cs b/src/CodeyBox.Api/Program.cs index 655981cc..d01b1a57 100644 --- a/src/CodeyBox.Api/Program.cs +++ b/src/CodeyBox.Api/Program.cs @@ -2061,6 +2061,10 @@ static IAgentQuotaProbe WireQuotaProbeTokenInvalidation( new QuotaReservationLedger( sp.GetRequiredService(), TimeProvider.System)); +builder.Services.AddSingleton(sp => + new ExecutorQuotaReportStore( + sp.GetRequiredService(), + TimeProvider.System)); builder.Services.AddSingleton(sp => { var cbOpts = sp.GetRequiredService>().Value; @@ -2096,7 +2100,8 @@ static IAgentQuotaProbe WireQuotaProbeTokenInvalidation( sp.GetService(), sp.GetRequiredService(), sp.GetService(), - sp.GetRequiredService()); + sp.GetRequiredService(), + sp.GetRequiredService()); }); // --- Per-agent concurrency / rate-aware dispatch ----------------------------- @@ -7252,6 +7257,14 @@ public sealed class QuotaRouterConfig /// Minutes observed quota failures are retained in state.db. Default 30. public int ObservedFailureRetentionMinutes { get; set; } = 30; /// + /// Tolerance in seconds for clock skew between executor and + /// orchestrator hosts when validating an executor-reported reading's + /// observed time. A report dated further in the future than this is + /// rejected. Default 300 (5 min). Hot-reloadable. + /// + public int ReportedReadingClockSkewSeconds { get; set; } = + QuotaRouterDefaults.DefaultReportedReadingClockSkewSeconds; + /// /// Seconds before a cap-spill-deferred work item is reconsidered (every /// eligible class member was at its per-agent concurrency cap). Default /// 15; the orchestrator's own atomic-reservation defer uses the same @@ -7415,6 +7428,33 @@ public sealed class QuotaPoolConfig /// reservation estimate. Set explicitly for balance pools. /// public double? ReservationEstimate { get; set; } + + /// + /// Where this pool's probe runs: OrchestratorDirect (the + /// default — the orchestrator holds the credential and probes + /// directly, exactly as today) or ExecutorReported (an + /// executor host holding the credential probes locally and reports + /// readings; the orchestrator meters the pool from the latest fresh + /// report). Case-insensitive. Hot-reloadable. + /// + public string ProbeSource { get; set; } = "OrchestratorDirect"; + + /// + /// Maximum age in seconds of an executor-reported reading before the + /// pool reads as unknown. Applies only to ExecutorReported + /// pools. Zero means the default (300); negative values are rejected. + /// Hot-reloadable. + /// + public int ReportedReadingMaxAgeSeconds { get; set; } = + QuotaRouterDefaults.DefaultReportedReadingMaxAgeSeconds; + + /// + /// Executor host ids authorised to report readings for this pool. + /// Applies only to ExecutorReported pools: reports from any + /// other host are rejected. Matched by exact equality against the + /// executor's registered host id. Hot-reloadable. + /// + public List HolderHostIds { get; set; } = []; } /// diff --git a/src/CodeyBox.Api/QuotaRouterConfigMapper.cs b/src/CodeyBox.Api/QuotaRouterConfigMapper.cs index 52eefc8c..8941a293 100644 --- a/src/CodeyBox.Api/QuotaRouterConfigMapper.cs +++ b/src/CodeyBox.Api/QuotaRouterConfigMapper.cs @@ -32,6 +32,10 @@ public static QuotaRouterOptions ToOptions(QuotaRouterConfig qr) UnknownPolicy = qr.UnknownPolicy, ObservedFailureWindow = TimeSpan.FromMinutes(qr.ObservedFailureWindowMinutes), ObservedFailureRetention = TimeSpan.FromMinutes(qr.ObservedFailureRetentionMinutes), + ReportedReadingClockSkew = qr.ReportedReadingClockSkewSeconds >= 0 + ? TimeSpan.FromSeconds(qr.ReportedReadingClockSkewSeconds) + : throw new InvalidOperationException( + "CodeyBox:QuotaRouter:ReportedReadingClockSkewSeconds must be >= 0."), CapRetryRecheckInterval = TimeSpan.FromSeconds(qr.CapRetryIntervalSeconds), ColdStartFitInWindow = qr.ColdStartFitInWindow, DrainAggressiveness = qr.DrainAggressiveness, @@ -75,6 +79,10 @@ public static void ApplyHotReload(QuotaRouterOptions dst, QuotaRouterConfig src) dst.UnknownPolicy = src.UnknownPolicy; dst.ObservedFailureWindow = TimeSpan.FromMinutes(src.ObservedFailureWindowMinutes); dst.ObservedFailureRetention = TimeSpan.FromMinutes(src.ObservedFailureRetentionMinutes); + if (src.ReportedReadingClockSkewSeconds < 0) + throw new InvalidOperationException( + "CodeyBox:QuotaRouter:ReportedReadingClockSkewSeconds must be >= 0."); + dst.ReportedReadingClockSkew = TimeSpan.FromSeconds(src.ReportedReadingClockSkewSeconds); dst.CapRetryRecheckInterval = TimeSpan.FromSeconds(src.CapRetryIntervalSeconds); dst.ColdStartFitInWindow = src.ColdStartFitInWindow; dst.DrainAggressiveness = src.DrainAggressiveness; @@ -86,6 +94,7 @@ public static void ApplyHotReload(QuotaRouterOptions dst, QuotaRouterConfig src) dst.QuotaReservationMaxAge = TimeSpan.FromSeconds(src.QuotaReservationMaxAgeSeconds); dst.ExpectedResets = BuildExpectedResetOverrides(src.ExpectedResets); dst.IntraKindRoutingPolicy = src.IntraKindRoutingPolicy; + QuotaPoolValidation.Validate(dst); } private static PausedQuotaMapping BuildPausedQuotaOptions(QuotaRouterConfig qr) @@ -205,11 +214,50 @@ private static Dictionary BuildPoolOptions( Kind = kind, BalanceUnit = string.IsNullOrWhiteSpace(kv.Value.BalanceUnit) ? null : kv.Value.BalanceUnit.Trim(), ReservationEstimate = estimate, + ProbeSource = ParseProbeSource(name, kv.Value.ProbeSource), + ReportedReadingMaxAge = kv.Value.ReportedReadingMaxAgeSeconds > 0 + ? TimeSpan.FromSeconds(kv.Value.ReportedReadingMaxAgeSeconds) + : kv.Value.ReportedReadingMaxAgeSeconds == 0 + ? QuotaRouterDefaults.DefaultReportedReadingMaxAge + : throw new InvalidOperationException( + $"Quota pool '{name}': ReportedReadingMaxAgeSeconds must be positive."), + HolderHostIds = BuildHolderHostIds(name, kv.Value.HolderHostIds), }; } return dst; } + private static QuotaProbeSource ParseProbeSource(string poolName, string? source) + { + if (string.IsNullOrWhiteSpace(source) + || string.Equals(source.Trim(), nameof(QuotaProbeSource.OrchestratorDirect), StringComparison.OrdinalIgnoreCase)) + return QuotaProbeSource.OrchestratorDirect; + if (string.Equals(source.Trim(), nameof(QuotaProbeSource.ExecutorReported), StringComparison.OrdinalIgnoreCase)) + return QuotaProbeSource.ExecutorReported; + throw new InvalidOperationException( + $"Quota pool '{poolName}': unknown probe source '{source}'. " + + $"Expected '{nameof(QuotaProbeSource.OrchestratorDirect)}' or '{nameof(QuotaProbeSource.ExecutorReported)}'."); + } + + private static List BuildHolderHostIds(string poolName, List? src) + { + var dst = new List(); + if (src is null) return dst; + foreach (var raw in src) + { + if (string.IsNullOrWhiteSpace(raw)) + throw new InvalidOperationException( + $"Quota pool '{poolName}' has an empty holder host id; " + + $"holder entries must name an executor host id."); + var hostId = raw.Trim(); + if (dst.Contains(hostId, StringComparer.Ordinal)) + throw new InvalidOperationException( + $"Quota pool '{poolName}' declares holder host id '{hostId}' more than once."); + dst.Add(hostId); + } + return dst; + } + private static QuotaPoolKind ParsePoolKind(string poolName, string? kind) { if (string.IsNullOrWhiteSpace(kind) diff --git a/src/CodeyBox.Orchestrator/AgentClassRouter.cs b/src/CodeyBox.Orchestrator/AgentClassRouter.cs index ccc85c14..927e6dc4 100644 --- a/src/CodeyBox.Orchestrator/AgentClassRouter.cs +++ b/src/CodeyBox.Orchestrator/AgentClassRouter.cs @@ -64,6 +64,7 @@ public sealed class AgentClassRouter : IAgentQuotaAvailabilitySnapshot, IAgentQu // estimated cost; the caller owns the returned lease and must release it // on the same lifecycle that releases the worker slot. private readonly QuotaReservationLedger? _reservationLedger; + private readonly ExecutorQuotaReportStore? _reportStore; private readonly IAgentQuotaAvailabilityPublisher? _quotaAvailabilityPublisher; private readonly AgentQuotaAvailabilityBroadcaster? _localQuotaAvailability; // Default fit when no historical samples exist (spec: "fits 2 concurrent @@ -114,7 +115,8 @@ public AgentClassRouter( IAgentDispatchAvailability? dispatchAvailability = null, IAgentQuotaAvailabilityPublisher? quotaAvailabilityPublisher = null, AgentCircuitBreaker? circuitBreaker = null, - QuotaReservationLedger? reservationLedger = null) + QuotaReservationLedger? reservationLedger = null, + ExecutorQuotaReportStore? reportStore = null) { _routingConfig = new RoutingConfig( catalog.ToDictionary(c => c.Id, StringComparer.OrdinalIgnoreCase), @@ -136,6 +138,7 @@ public AgentClassRouter( _dispatchAvailability = dispatchAvailability; _circuitBreaker = circuitBreaker; _reservationLedger = reservationLedger; + _reportStore = reportStore; _quotaAvailabilityPublisher = quotaAvailabilityPublisher; if (quotaAvailabilityPublisher is not IAgentQuotaAvailabilitySignal) _localQuotaAvailability = new AgentQuotaAvailabilityBroadcaster(); @@ -1954,24 +1957,52 @@ private async Task ProbeOrUnknownAsync(AgentMembership membe /// pool-mates within a dispatch pass: the first member evaluated for a /// pool performs the probe and later members reuse the identical snapshot /// from . Members with no (or an unresolvable) - /// pool always probe directly. The probe itself is unchanged — only the - /// redundant call is skipped. + /// pool always probe directly. Members of an executor-reported pool never + /// probe directly — the orchestrator holds no credential for those + /// accounts — and instead meter from the latest fresh executor report; + /// a stale or missing report reads as unknown. The probe itself is + /// unchanged — only the redundant call is skipped. /// private async Task ProbePoolMemberAsync( AgentMembership member, Dictionary poolCache, CancellationToken ct) { - if (QuotaPoolResolver.TryResolvePool(_opts, member, out var poolName, out _, out _) + string? poolName = null; + QuotaPoolOptions? pool = null; + if (QuotaPoolResolver.TryResolvePool(_opts, member, out poolName, out pool, out _) && poolName is not null && poolCache.TryGetValue(poolName, out var cached)) return cached; - var snapshot = await ProbeOrUnknownAsync(member, ct); + AgentQuotaSnapshot snapshot; + if (pool is not null + && poolName is not null + && pool.ProbeSource == QuotaProbeSource.ExecutorReported) + snapshot = ReadExecutorReportedSnapshot(poolName); + else + snapshot = await ProbeOrUnknownAsync(member, ct); if (poolName is not null) poolCache[poolName] = snapshot; return snapshot; } + /// + /// Serves the snapshot for an executor-reported pool from the report + /// store. No direct probe is attempted — there is no orchestrator-held + /// credential to probe against — so a missing store reads as unknown and + /// the gate's standard unknown handling applies. Admission is still + /// decided by on the orchestrator; this + /// only supplies the reading. + /// + private AgentQuotaSnapshot ReadExecutorReportedSnapshot(string poolName) + { + if (_reportStore is null) + return AgentQuotaSnapshot.UnknownSnapshot( + QuotaUnknownReason.Transient, + $"quota pool '{poolName}' is executor-reported but no report store is wired"); + return _reportStore.GetSnapshot(poolName); + } + /// /// The ledger-facing reading for a probe snapshot in the member's native /// pool unit: the absolute balance for depleting-balance pools (their @@ -3360,6 +3391,15 @@ public sealed class QuotaRouterOptions public TimeSpan ObservedFailureRetention { get; set; } = TimeSpan.FromMinutes(30); + /// + /// Tolerance for clock skew between executor and orchestrator hosts when + /// validating an executor-reported reading's observed time. A report dated + /// further in the future than this is rejected. Default 5 minutes. + /// Hot-reloadable. + /// + public TimeSpan ReportedReadingClockSkew { get; set; } = + QuotaRouterDefaults.DefaultReportedReadingClockSkew; + /// /// Suggested recheck delay surfaced by /// when every eligible candidate was blocked by its per-agent concurrency diff --git a/src/CodeyBox.Orchestrator/ExecutorQuotaReportStore.cs b/src/CodeyBox.Orchestrator/ExecutorQuotaReportStore.cs new file mode 100644 index 00000000..73868bde --- /dev/null +++ b/src/CodeyBox.Orchestrator/ExecutorQuotaReportStore.cs @@ -0,0 +1,429 @@ +using CodeyBox.Core; + +namespace CodeyBox.Orchestrator; + +/// +/// A quota reading observed by the executor host holding a pool's credential, +/// reported to the orchestrator. Carries the pool identity, the availability +/// reading, the reset time, and the time it was observed — the four facts the +/// gate needs. It never carries an admission verdict: the orchestrator remains +/// the sole authority for the gate decision and re-evaluates every stored +/// reading through on each dispatch. +/// +public sealed record ExecutorQuotaReport +{ + /// Pool this reading meters, matching a configured pool name (case-insensitive). + public required string PoolName { get; init; } + + /// + /// Percentage of quota remaining (0–100) for resetting-window pools. + /// Required for a known resetting-window reading; informational otherwise. + /// + public double? AvailablePct { get; init; } + + /// + /// Absolute remaining balance for depleting-balance pools, in the pool's + /// native unit. Required for a known depleting-balance reading. + /// + public double? BalanceRemaining { get; init; } + + /// + /// When the quota window resets, if known. Meaningful only for + /// resetting-window pools — a depleting-balance pool never resets, so a + /// report carrying one for such a pool is rejected rather than stored. + /// + public DateTimeOffset? ResetAt { get; init; } + + /// + /// When the executor observed this reading. Freshness (and therefore + /// whether the pool meters from this report or reads as unknown) is + /// evaluated against this instant, not arrival time. + /// + public required DateTimeOffset ObservedAt { get; init; } + + /// + /// Set when the executor could produce no real reading: its local probe's + /// unknown reason, preserved so the orchestrator applies the same unknown + /// handling as a direct probe ( + /// retains recent-good data, / + /// discard it). Null means + /// this report carries a real reading. + /// + public QuotaUnknownReason? Unknown { get; init; } + + /// Human-readable notes, e.g. the probe endpoint outcome. Bounded on arrival. + public string? Notes { get; init; } +} + +/// +/// Orchestrator-side store for executor-reported quota readings. Executors +/// holding a pool's credential probe locally and POST their readings here; +/// the quota gate meters pools +/// from the latest fresh report instead of probing directly (the orchestrator +/// holds no credential for those accounts and must not probe against nothing). +/// +/// Trust boundary: every report is validated at this sink before it is +/// stored. The pool must exist and be executor-reported, the reporting host +/// must be operator-declared in the pool's HolderHostIds (exact ordinal +/// match — a misconfigured or compromised executor cannot overwrite a meter +/// it does not own), numeric readings must be in range, and the reset time +/// must be consistent with the pool's kind. Rejected reports leave the stored +/// reading unchanged. +/// +/// Freshness is evaluated at read time against the pool's +/// ReportedReadingMaxAge: a stale or missing report reads as an unknown +/// snapshot (never as healthy headroom) and flows through the same unknown +/// handling as a direct probe, including fail-closed whenever a non-zero floor +/// is in force. This store never decides admission — it only accepts or +/// rejects reports and serves snapshots to the gate. +/// +/// Thread-safe. The options reference is the live shared instance, so +/// hot-reload edits to probe source, staleness bound, and holder allowlists +/// take effect on the next report or read without a restart. +/// +public sealed class ExecutorQuotaReportStore +{ + /// Maximum note length accepted on a report; longer notes are rejected, not truncated. + public const int MaxReportNotesLength = 512; + + private readonly QuotaRouterOptions _options; + private readonly TimeProvider _time; + private readonly object _lock = new(); + private readonly Dictionary _readings = new(StringComparer.OrdinalIgnoreCase); + + private sealed record StoredReport(ExecutorQuotaReport Report, string ReportedByHostId); + + public ExecutorQuotaReportStore(QuotaRouterOptions options, TimeProvider? timeProvider = null) + { + _options = options ?? throw new ArgumentNullException(nameof(options)); + _time = timeProvider ?? TimeProvider.System; + } + + /// + /// Validates from and, + /// when valid, stores it as the pool's current reading (replacing any + /// prior report). Returns true with a null reason on acceptance; false + /// with a human-readable reason on rejection. Rejection never mutates the + /// stored reading. The outcome is a storage verdict only — never a gate + /// admission decision, which stays with on + /// the orchestrator. + /// + public bool TryReport(string? hostId, ExecutorQuotaReport? report, out string? rejectionReason) + { + var now = _time.GetUtcNow(); + if (!TryValidate(hostId, report, now, out var poolName, out var pool, out var normalized, out rejectionReason)) + return false; + + lock (_lock) + _readings[poolName!] = new StoredReport(normalized!, hostId!.Trim()); + rejectionReason = null; + return true; + } + + /// + /// Serves the pool's current snapshot for the gate: the stored reading + /// while it is fresh, otherwise an unknown snapshot. A missing report, a + /// report older than the pool's ReportedReadingMaxAge, or a pool + /// that is not executor-reported all read as + /// unknown so the gate's + /// standard unknown handling (including fail-closed on a non-zero floor) + /// applies and silence never presents as headroom. A fresh explicitly + /// unknown report preserves its so + /// retain-vs-discard semantics match a direct probe. + /// + public AgentQuotaSnapshot GetSnapshot(string? poolName) + { + var now = _time.GetUtcNow(); + var normalized = QuotaPoolResolver.NormalizePoolName(poolName); + if (normalized is null) + return AgentQuotaSnapshot.UnknownSnapshot( + QuotaUnknownReason.Transient, "executor-reported quota read with no pool name"); + if (_options.Pools is not { } pools + || !pools.TryGetValue(normalized, out var pool) + || pool is null) + return AgentQuotaSnapshot.UnknownSnapshot( + QuotaUnknownReason.Transient, $"no configured quota pool '{normalized}'"); + if (pool.ProbeSource != QuotaProbeSource.ExecutorReported) + return AgentQuotaSnapshot.UnknownSnapshot( + QuotaUnknownReason.Transient, + $"quota pool '{normalized}' is orchestrator-probed; no executor reading applies"); + + StoredReport? stored; + lock (_lock) + _readings.TryGetValue(normalized, out stored); + if (stored is null) + return AgentQuotaSnapshot.UnknownSnapshot( + QuotaUnknownReason.Transient, + $"no executor reading reported for pool '{normalized}'"); + + var age = now - stored.Report.ObservedAt; + if (age < TimeSpan.Zero) + age = TimeSpan.Zero; + if (age > pool.ReportedReadingMaxAge) + return AgentQuotaSnapshot.UnknownSnapshot( + QuotaUnknownReason.Transient, + $"executor reading for pool '{normalized}' is stale " + + $"(age {(long)Math.Round(age.TotalSeconds)}s > bound {(long)Math.Round(pool.ReportedReadingMaxAge.TotalSeconds)}s)"); + + if (stored.Report.Unknown is { } reason) + return AgentQuotaSnapshot.UnknownSnapshot( + reason, + $"executor '{stored.ReportedByHostId}' reported unknown for pool '{normalized}'" + + (string.IsNullOrWhiteSpace(stored.Report.Notes) ? "" : $" ({stored.Report.Notes})")); + + var ageSeconds = (long)Math.Round(age.TotalSeconds); + var balanceUnit = string.IsNullOrWhiteSpace(pool.BalanceUnit) ? null : pool.BalanceUnit.Trim(); + var noteDetail = string.IsNullOrWhiteSpace(stored.Report.Notes) ? "" : $" ({stored.Report.Notes})"; + if (pool.Kind == QuotaPoolKind.DepletingBalance) + return new AgentQuotaSnapshot + { + AvailablePct = stored.Report.AvailablePct ?? -1, + BalanceRemaining = stored.Report.BalanceRemaining, + BalanceUnit = balanceUnit, + Notes = $"executor '{stored.ReportedByHostId}' reading (age {ageSeconds}s){noteDetail}", + }; + if (stored.Report.AvailablePct is not { } pct) + return AgentQuotaSnapshot.UnknownSnapshot( + QuotaUnknownReason.Transient, + $"executor reading for pool '{normalized}' carries no percentage " + + "(pool reconfigured since the report was stored)"); + return new AgentQuotaSnapshot + { + AvailablePct = pct, + ResetAt = stored.Report.ResetAt, + BalanceRemaining = stored.Report.BalanceRemaining, + BalanceUnit = balanceUnit, + Notes = $"executor '{stored.ReportedByHostId}' reading (age {ageSeconds}s){noteDetail}", + }; + } + + /// + /// Returns the currently stored report for , + /// regardless of freshness, for diagnostics and tests. Freshness is a + /// read-time concern (see ); this accessor does + /// not apply it. + /// + public bool TryGetStored(string? poolName, out ExecutorQuotaReport? report, out string? reportedByHostId) + { + report = null; + reportedByHostId = null; + var normalized = QuotaPoolResolver.NormalizePoolName(poolName); + if (normalized is null) + return false; + lock (_lock) + { + if (!_readings.TryGetValue(normalized, out var stored)) + return false; + report = stored.Report; + reportedByHostId = stored.ReportedByHostId; + return true; + } + } + + private bool TryValidate( + string? hostId, + ExecutorQuotaReport? report, + DateTimeOffset now, + out string? poolName, + out QuotaPoolOptions? pool, + out ExecutorQuotaReport? normalized, + out string? rejectionReason) + { + poolName = null; + pool = null; + normalized = null; + rejectionReason = null; + + var host = string.IsNullOrWhiteSpace(hostId) ? null : hostId.Trim(); + if (host is null) + { + rejectionReason = "executor report rejected: reporting host id is required."; + return false; + } + if (report is null) + { + rejectionReason = "executor report rejected: report body is required."; + return false; + } + + poolName = QuotaPoolResolver.NormalizePoolName(report.PoolName); + if (poolName is null) + { + rejectionReason = "executor report rejected: pool name is required."; + return false; + } + if (_options.Pools is not { } pools + || !pools.TryGetValue(poolName, out pool) + || pool is null) + { + pool = null; + rejectionReason = $"executor report rejected: no configured quota pool '{poolName}'."; + return false; + } + if (pool.ProbeSource != QuotaProbeSource.ExecutorReported) + { + rejectionReason = + $"executor report rejected: quota pool '{poolName}' is orchestrator-probed; " + + "reports are accepted only for executor-reported pools."; + return false; + } + if (!HoldsPool(pool, host)) + { + rejectionReason = + $"executor report rejected: host '{host}' is not declared as holding quota pool '{poolName}'."; + return false; + } + if (report.ObservedAt == default) + { + rejectionReason = $"executor report rejected for pool '{poolName}': observed time is required."; + return false; + } + if (report.ObservedAt > now + _options.ReportedReadingClockSkew) + { + rejectionReason = $"executor report rejected for pool '{poolName}': observed time is in the future."; + return false; + } + if (report.Notes is { Length: > MaxReportNotesLength }) + { + rejectionReason = + $"executor report rejected for pool '{poolName}': notes exceed {MaxReportNotesLength} characters."; + return false; + } + if (report.Notes is { } notes && notes.Any(char.IsControl)) + { + rejectionReason = + $"executor report rejected for pool '{poolName}': notes must not contain control characters."; + return false; + } + if (report.Unknown is { } unknown && !Enum.IsDefined(unknown)) + { + rejectionReason = $"executor report rejected for pool '{poolName}': unknown reason is not recognised."; + return false; + } + if (ValidateReading(pool, report, now, _options.ReportedReadingClockSkew, MaxResetHorizon(_options)) is { } readingRejection) + { + rejectionReason = readingRejection; + return false; + } + + normalized = report with { PoolName = poolName }; + return true; + } + + /// + /// Validates the reading carried by a report against its pool's kind: + /// percentages within 0-100, balances finite and non-negative, no reset + /// instant on a depleting-balance pool (which never resets), and — for a + /// resetting-window pool carrying a reset — a reset after the observed + /// instant (within clock skew) and inside the plausible horizon (the + /// widest configured ramp window plus skew). An unbounded executor-set + /// reset would otherwise pin the gate's ramped floor at one end of its + /// range or surface a bogus retry hint. Returns null when the reading is + /// acceptable, otherwise the rejection reason. Pure. + /// + private static string? ValidateReading( + QuotaPoolOptions pool, + ExecutorQuotaReport report, + DateTimeOffset now, + TimeSpan clockSkew, + TimeSpan resetHorizon) + { + var poolName = pool.Name; + if (pool.Kind == QuotaPoolKind.DepletingBalance) + { + if (report.ResetAt is not null) + return $"executor report rejected for pool '{poolName}': a depleting-balance pool " + + "never carries a reset instant."; + if (report.Unknown is null) + { + if (report.BalanceRemaining is not { } balance + || !double.IsFinite(balance) + || balance < 0) + return $"executor report rejected for pool '{poolName}': balance must be " + + "a finite non-negative value."; + if (report.AvailablePct is { } pct && (!double.IsFinite(pct) || pct is < 0 or > 100)) + return $"executor report rejected for pool '{poolName}': percentage must be " + + "within 0-100 when present."; + return null; + } + return ValidateAccompanyingValues(poolName, report); + } + + if (report.ResetAt is { } reset) + { + if (reset <= report.ObservedAt - clockSkew) + return $"executor report rejected for pool '{poolName}': reset must be " + + "after the time the reading was observed."; + if (reset > now + resetHorizon + clockSkew) + return $"executor report rejected for pool '{poolName}': reset is " + + "beyond the plausible horizon for this pool."; + } + if (report.Unknown is null) + { + if (report.AvailablePct is not { } pct || !double.IsFinite(pct) || pct is < 0 or > 100) + return $"executor report rejected for pool '{poolName}': available percentage " + + "must be within 0-100."; + if (!NonNegativeOrAbsent(report.BalanceRemaining)) + return $"executor report rejected for pool '{poolName}': balance must be " + + "a finite non-negative value when present."; + return null; + } + return ValidateAccompanyingValues(poolName, report); + } + + private static string? ValidateAccompanyingValues(string poolName, ExecutorQuotaReport report) => + RangedOrAbsent(report.AvailablePct, 0, 100) + && NonNegativeOrAbsent(report.BalanceRemaining) + ? null + : $"executor report rejected for pool '{poolName}': accompanying values " + + "are outside their valid ranges."; + + /// + /// Widest ramp window configured anywhere (global, per-agent, per-pool), + /// bounding how far ahead an executor-reported reset may lie. A reset + /// beyond this horizon could never key a live ramp, so it is rejected as + /// inconsistent rather than stored. Pure. + /// + private static TimeSpan MaxResetHorizon(QuotaRouterOptions options) + { + var horizon = options.RampWindow; + if (options.RampWindowByAgent is { } byAgent) + { + foreach (var window in byAgent.Values) + { + if (window > horizon) + horizon = window; + } + } + if (options.FloorByPool is { } floors) + { + foreach (var floor in floors.Values) + { + if (floor?.RampWindow is { } window && window > horizon) + horizon = window; + } + } + return horizon > TimeSpan.Zero ? horizon : QuotaRouterDefaults.DefaultRampWindow; + } + + /// + /// True when is operator-declared as holding the + /// pool. Exact ordinal equality — never substring — so "exec-1" never + /// implies "exec-10". + /// + private static bool HoldsPool(QuotaPoolOptions pool, string hostId) + { + foreach (var entry in pool.HolderHostIds) + { + if (string.Equals(entry?.Trim(), hostId, StringComparison.Ordinal)) + return true; + } + return false; + } + + private static bool RangedOrAbsent(double? value, double min, double max) => + value is not { } v || (double.IsFinite(v) && v >= min && v <= max); + + private static bool NonNegativeOrAbsent(double? value) => + value is not { } v || (double.IsFinite(v) && v >= 0); +} diff --git a/src/CodeyBox.Orchestrator/QuotaPools.cs b/src/CodeyBox.Orchestrator/QuotaPools.cs index 3e81ffe2..f88427ff 100644 --- a/src/CodeyBox.Orchestrator/QuotaPools.cs +++ b/src/CodeyBox.Orchestrator/QuotaPools.cs @@ -29,6 +29,34 @@ public enum QuotaPoolKind DepletingBalance, } +/// +/// Where a quota pool's probe runs and how its reading reaches the gate. +/// The default keeps current behaviour: the orchestrator holds the credential +/// and probes directly in-process. The executor-reported mode is for accounts +/// whose credential lives on an executor host the orchestrator cannot read: +/// that host probes locally and reports readings, and the orchestrator meters +/// the pool from the latest fresh reported reading. The gate decision itself +/// is always made by the orchestrator in both modes — an executor reports +/// readings and never decides admission. +/// +public enum QuotaProbeSource +{ + /// + /// The orchestrator holds the credential and probes directly in-process. + /// Current behaviour; the default for every pool. + /// + OrchestratorDirect, + + /// + /// An executor host holding the credential probes locally and reports + /// readings to the orchestrator, which meters the pool from the latest + /// fresh report. A stale or missing report reads as unknown (never as + /// healthy headroom); reports from hosts not declared in + /// are rejected. + /// + ExecutorReported, +} + /// /// Operator-declared identity for one underlying account or subscription. /// Pool membership is declared per class member @@ -64,6 +92,38 @@ public sealed class QuotaPoolOptions /// Hot-reloadable. /// public double? ReservationEstimate { get; set; } + + /// + /// Where this pool's probe runs. + /// (the default) probes in-process against orchestrator-held credentials, + /// exactly as today. meters + /// the pool from readings reported by the executor host(s) holding the + /// credential. Hot-reloadable. + /// + public QuotaProbeSource ProbeSource { get; set; } = QuotaProbeSource.OrchestratorDirect; + + /// + /// Maximum age of an executor-reported reading before the pool reads as + /// unknown. Applies only to + /// pools; the unknown then flows through the same unknown handling as a + /// direct probe (fail-closed whenever a non-zero floor is in force). + /// Silence from an executor therefore never presents as healthy headroom. + /// Hot-reloadable. Must be positive; defaults to + /// . + /// + public TimeSpan ReportedReadingMaxAge { get; set; } = QuotaRouterDefaults.DefaultReportedReadingMaxAge; + + /// + /// Executor host ids authorised to report readings for this pool. + /// Applies only to pools: + /// a report for the pool from any other host is rejected and the stored + /// reading is left unchanged, so a misconfigured or compromised executor + /// cannot overwrite a meter it does not own. Matched by exact ordinal + /// equality against the executor's registered host id — never by + /// substring. Operator-declared; never self-asserted by the executor. + /// Hot-reloadable. + /// + public List HolderHostIds { get; set; } = []; } /// @@ -226,6 +286,9 @@ public static class QuotaPoolValidation public static void Validate(QuotaRouterOptions options) { ArgumentNullException.ThrowIfNull(options); + if (options.ReportedReadingClockSkew < TimeSpan.Zero) + throw new InvalidOperationException( + $"ReportedReadingClockSkew ({options.ReportedReadingClockSkew}) must be >= 0."); if (options.Pools is { } pools) { foreach (var (key, pool) in pools) @@ -233,6 +296,15 @@ public static void Validate(QuotaRouterOptions options) if (pool is null) throw new InvalidOperationException( $"Quota pool '{key}' has no configuration; declare its replenishment kind."); + if (!Enum.IsDefined(pool.ProbeSource)) + throw new InvalidOperationException( + $"Quota pool '{key}' names an unknown probe source '{(int)pool.ProbeSource}'; " + + $"expected '{nameof(QuotaProbeSource.OrchestratorDirect)}' or '{nameof(QuotaProbeSource.ExecutorReported)}'."); + if (pool.ReportedReadingMaxAge <= TimeSpan.Zero) + throw new InvalidOperationException( + $"Quota pool '{key}' must have a positive ReportedReadingMaxAge; " + + $"staleness without a bound would present silence as headroom."); + ValidateHolderHostIds(key, pool); } } if (options.FloorByPool is not { } floors) @@ -269,4 +341,34 @@ public static void Validate(QuotaRouterOptions options) } } } + + /// + /// Validates the operator-declared holder allowlist for one pool: entries + /// must be non-empty, bounded like executor host ids, and free of control + /// characters. An executor-reported pool with no holders accepts no + /// reports, so it meters as unknown until the operator declares who holds + /// the credential — that gap fails closed at the gate rather than at load. + /// + private static void ValidateHolderHostIds(string poolName, QuotaPoolOptions pool) + { + var seen = new HashSet(StringComparer.Ordinal); + foreach (var raw in pool.HolderHostIds) + { + if (string.IsNullOrWhiteSpace(raw)) + throw new InvalidOperationException( + $"Quota pool '{poolName}' has an empty holder host id; " + + $"holder entries must name an executor host id."); + var hostId = raw.Trim(); + if (hostId.Length > CodeyBox.Core.ExecutorRegistration.MaxHostIdLength) + throw new InvalidOperationException( + $"Quota pool '{poolName}' holder host id '{hostId}' exceeds " + + $"{CodeyBox.Core.ExecutorRegistration.MaxHostIdLength} characters."); + if (hostId.Any(char.IsControl)) + throw new InvalidOperationException( + $"Quota pool '{poolName}' holder host id must not contain control characters."); + if (!seen.Add(hostId)) + throw new InvalidOperationException( + $"Quota pool '{poolName}' declares holder host id '{hostId}' more than once."); + } + } } diff --git a/src/CodeyBox.Orchestrator/QuotaRouterDefaults.cs b/src/CodeyBox.Orchestrator/QuotaRouterDefaults.cs index 28c0dd95..ddbc6191 100644 --- a/src/CodeyBox.Orchestrator/QuotaRouterDefaults.cs +++ b/src/CodeyBox.Orchestrator/QuotaRouterDefaults.cs @@ -6,6 +6,7 @@ public static class QuotaRouterDefaults public const int DefaultQuotaRecoveryProbeIntervalSeconds = 5; public const int DefaultQuotaRecoveryProbeEligibilityScanLimit = 128; public const int DefaultQuotaReservationMaxAgeSeconds = 6 * 60 * 60; + public const int DefaultReportedReadingMaxAgeSeconds = 5 * 60; public static TimeSpan DefaultRampWindow { get; } = TimeSpan.FromSeconds(DefaultRampWindowSeconds); @@ -15,4 +16,23 @@ public static class QuotaRouterDefaults public static TimeSpan DefaultQuotaReservationMaxAge { get; } = TimeSpan.FromSeconds(DefaultQuotaReservationMaxAgeSeconds); + + public const int DefaultReportedReadingClockSkewSeconds = 5 * 60; + + /// + /// Tolerance for clock skew between executor and orchestrator hosts when + /// validating an executor-reported reading's observed time. A report dated + /// further in the future than this is rejected (a future-dated report + /// would otherwise read as fresh for longer than its bound allows). + /// + public static TimeSpan DefaultReportedReadingClockSkew { get; } = + TimeSpan.FromSeconds(DefaultReportedReadingClockSkewSeconds); + + /// + /// Default maximum age of an executor-reported quota reading before the + /// pool reads as unknown. Matches the last-known-good retention horizon + /// so executor silence and probe silence age out on the same cadence. + /// + public static TimeSpan DefaultReportedReadingMaxAge { get; } = + TimeSpan.FromSeconds(DefaultReportedReadingMaxAgeSeconds); } diff --git a/tests/CodeyBox.Tests/ExecutorQuotaReportAuthTests.cs b/tests/CodeyBox.Tests/ExecutorQuotaReportAuthTests.cs new file mode 100644 index 00000000..61443768 --- /dev/null +++ b/tests/CodeyBox.Tests/ExecutorQuotaReportAuthTests.cs @@ -0,0 +1,389 @@ +using System.Net; +using System.Net.Http.Json; +using System.Text.Json; +using CodeyBox.Api; +using CodeyBox.Core; +using CodeyBox.Orchestrator; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Mvc.Testing; +using Microsoft.AspNetCore.TestHost; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Hosting; + +namespace CodeyBox.Tests; + +/// +/// Caller binding for executor quota-report ingress: the path host must +/// match the authenticated caller's host-bound token. A shared bearer (the +/// operator key or any token without an ExecutorHostId binding) +/// proves nothing about which host is calling, so it cannot report readings +/// — otherwise any bearer holder could forge another pool's meter by +/// asserting a victim host id. The pool's HolderHostIds remain the +/// second check inside the store. +/// +public sealed class ExecutorQuotaReportCallerTests +{ + private static WorkInitiator Initiator(string subject) => new() + { + Issuer = "test", + Subject = subject, + DisplayName = subject, + }; + + private static DefaultHttpContext ContextWith(ApiClientPrincipal principal) + { + var ctx = new DefaultHttpContext(); + ctx.Items[ApiKeyAuth.PrincipalItemKey] = principal; + return ctx; + } + + private static int GetStatusCode(IResult result) + { + var status = Assert.IsAssignableFrom(result); + return status.StatusCode ?? StatusCodes.Status200OK; + } + + [Fact] + public void MissingPrincipal_IsUnauthorized() + { + var result = ExecutorEndpoints.CheckQuotaReportCaller(new DefaultHttpContext(), "exec-1"); + + Assert.NotNull(result); + } + + [Fact] + public void MissingPrincipal_Returns401() + { + var result = ExecutorEndpoints.CheckQuotaReportCaller(new DefaultHttpContext(), "exec-1"); + + Assert.Equal(StatusCodes.Status401Unauthorized, GetStatusCode(result!)); + } + + [Fact] + public void AuthenticationDisabled_IsAllowed() + { + var ctx = ContextWith(new ApiClientPrincipal( + ApiKeyAuth.AuthenticationDisabledClientName, + Initiator("operator"), + CanDelegateInitiator: false)); + + Assert.Null(ExecutorEndpoints.CheckQuotaReportCaller(ctx, "exec-1")); + } + + [Fact] + public void UnboundToken_IsForbidden() + { + // The shared operator key and any named token without an + // ExecutorHostId binding carry no host identity: rejecting them here + // is what stops one bearer holder forging another host's meter. + var ctx = ContextWith(new ApiClientPrincipal( + "legacy-operator", + Initiator("operator"), + CanDelegateInitiator: false)); + + var result = ExecutorEndpoints.CheckQuotaReportCaller(ctx, "exec-1"); + + Assert.NotNull(result); + Assert.Equal(StatusCodes.Status403Forbidden, GetStatusCode(result)); + } + + [Fact] + public void BoundToken_MatchingHost_IsAllowed() + { + var ctx = ContextWith(new ApiClientPrincipal( + "exec-1-client", + Initiator("exec-1"), + CanDelegateInitiator: false, + ExecutorHostId: "exec-1")); + + Assert.Null(ExecutorEndpoints.CheckQuotaReportCaller(ctx, "exec-1")); + } + + [Fact] + public void BoundToken_OtherHost_IsForbidden() + { + var ctx = ContextWith(new ApiClientPrincipal( + "exec-2-client", + Initiator("exec-2"), + CanDelegateInitiator: false, + ExecutorHostId: "exec-2")); + + var result = ExecutorEndpoints.CheckQuotaReportCaller(ctx, "exec-1"); + + Assert.NotNull(result); + Assert.Equal(StatusCodes.Status403Forbidden, GetStatusCode(result)); + } + + [Fact] + public void BoundToken_HostIdComparison_IsExactOrdinal() + { + var ctx = ContextWith(new ApiClientPrincipal( + "exec-1-client", + Initiator("exec-1"), + CanDelegateInitiator: false, + ExecutorHostId: "Exec-1")); + + var result = ExecutorEndpoints.CheckQuotaReportCaller(ctx, "exec-1"); + + Assert.NotNull(result); + Assert.Equal(StatusCodes.Status403Forbidden, GetStatusCode(result)); + } +} + +/// +/// HTTP-level wiring for quota-report ingress: the endpoint's own layer +/// (live registry registration) composed with the real +/// (holder allowlist, storage). +/// Runs with authentication disabled so the caller-binding layer above +/// passes through as the local operator. +/// +[Collection("GlobalSerilog")] +public sealed class ExecutorQuotaReportIngressTests : IDisposable +{ + private readonly QuotaReportIngressFactory _factory = new(); + private readonly HttpClient _client; + + public ExecutorQuotaReportIngressTests() + { + _client = _factory.CreateClient(); + } + + public void Dispose() + { + _client.Dispose(); + _factory.Dispose(); + } + + private async Task RegisterAsync(string hostId) + { + var resp = await _client.PostAsJsonAsync( + "/executors/register", new { hostId }); + Assert.Equal(HttpStatusCode.OK, resp.StatusCode); + } + + private static object ReportBody(string pool, double availablePct) => new + { + pool, + availablePct, + observedAt = DateTimeOffset.UtcNow, + }; + + [Fact] + public async Task UnregisteredHost_ReturnsNotFound() + { + var resp = await _client.PostAsJsonAsync( + "/executors/ghost/quota-reports", ReportBody("exec-pool", 42.5)); + + Assert.Equal(HttpStatusCode.NotFound, resp.StatusCode); + } + + [Fact] + public async Task NonHolderHost_ReturnsBadRequest() + { + await RegisterAsync("exec-1"); + await RegisterAsync("exec-2"); + + var resp = await _client.PostAsJsonAsync( + "/executors/exec-2/quota-reports", ReportBody("exec-pool", 42.5)); + + Assert.Equal(HttpStatusCode.BadRequest, resp.StatusCode); + } + + [Fact] + public async Task DeclaredHolder_ReturnsOkAndStoresReading() + { + await RegisterAsync("exec-1"); + + var resp = await _client.PostAsJsonAsync( + "/executors/exec-1/quota-reports", ReportBody("exec-pool", 42.5)); + + Assert.Equal(HttpStatusCode.OK, resp.StatusCode); + var body = await resp.Content.ReadFromJsonAsync(); + Assert.True(body.GetProperty("accepted").GetBoolean()); + Assert.Equal("exec-pool", body.GetProperty("pool").GetString()); + + var store = _factory.Services.GetRequiredService(); + var snapshot = store.GetSnapshot("exec-pool"); + Assert.Equal(42.5, snapshot.AvailablePct); + } +} + +internal sealed class QuotaReportIngressFactory : WebApplicationFactory +{ + private readonly string _dbPath = Path.Combine( + Path.GetTempPath(), $"codeybox-quota-ingress-{Guid.NewGuid():N}.db"); + + protected override void ConfigureWebHost(IWebHostBuilder builder) + { + builder.UseEnvironment("Development"); + builder.ConfigureAppConfiguration((_, cfg) => + { + var tmp = Path.GetTempPath(); + cfg.AddInMemoryCollection(new Dictionary + { + ["CodeyBox:DangerouslyDisableAuth"] = "true", + ["CodeyBox:StateDatabasePath"] = _dbPath, + ["CodeyBox:GitRootDirectory"] = Path.Combine(tmp, $"test-git-{Guid.NewGuid():N}"), + ["CodeyBox:AuditLog:Path"] = Path.Combine(tmp, $"test-log-{Guid.NewGuid():N}-.json"), + ["CodeyBox:AuditLog:AuditPath"] = Path.Combine(tmp, $"test-audit-{Guid.NewGuid():N}-.json"), + ["CodeyBox:QuotaRouter:Pools:exec-pool:Kind"] = "ResettingWindow", + ["CodeyBox:QuotaRouter:Pools:exec-pool:ProbeSource"] = "ExecutorReported", + ["CodeyBox:QuotaRouter:Pools:exec-pool:ReportedReadingMaxAgeSeconds"] = "300", + ["CodeyBox:QuotaRouter:Pools:exec-pool:HolderHostIds:0"] = "exec-1", + }); + }); + builder.ConfigureTestServices(services => + { + services.RemoveAll(); + }); + } + + protected override void Dispose(bool disposing) + { + if (disposing) + { + try { File.Delete(_dbPath); } catch { } + } + base.Dispose(disposing); + } +} + +/// +/// End-to-end caller binding through the real bearer middleware: requests +/// carrying a host-bound executor token, an unbound token, or no token at +/// all are admitted or rejected by the token-to-host binding before the +/// registry is even consulted. The singleton is +/// replaced in DI, so no process environment variable is touched. +/// +[Collection("GlobalSerilog")] +public sealed class ExecutorQuotaReportMiddlewareTests : IDisposable +{ + private const string ExecutorToken = "test-bearer-bound-to-exec-1"; + private const string UnboundToken = "test-bearer-with-no-host-binding"; + + private readonly QuotaReportAuthFactory _factory = new(); + + public void Dispose() => _factory.Dispose(); + + private HttpClient ClientWith(string? bearer) + { + var client = _factory.CreateClient(); + if (bearer is not null) + client.DefaultRequestHeaders.Authorization = + new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", bearer); + return client; + } + + [Fact] + public async Task BoundToken_MatchingHost_ReportsSuccessfully() + { + using var client = ClientWith(ExecutorToken); + + var registered = await client.PostAsJsonAsync("/executors/register", new { hostId = "exec-1" }); + Assert.Equal(HttpStatusCode.OK, registered.StatusCode); + var reported = await client.PostAsJsonAsync( + "/executors/exec-1/quota-reports", + new { pool = "exec-pool", availablePct = 55.0, observedAt = DateTimeOffset.UtcNow }); + + Assert.Equal(HttpStatusCode.OK, reported.StatusCode); + } + + [Fact] + public async Task BoundToken_OtherHost_IsRejectedBeforeRegistryLookup() + { + using var client = ClientWith(ExecutorToken); + + // exec-2 is never registered: a 403 (not 404) proves the caller + // binding runs before the registry existence check, so a rejected + // caller cannot probe which host ids exist. + var reported = await client.PostAsJsonAsync( + "/executors/exec-2/quota-reports", + new { pool = "exec-pool", availablePct = 55.0, observedAt = DateTimeOffset.UtcNow }); + + Assert.Equal(HttpStatusCode.Forbidden, reported.StatusCode); + } + + [Fact] + public async Task UnboundToken_IsRejected() + { + using var client = ClientWith(UnboundToken); + + var registered = await client.PostAsJsonAsync("/executors/register", new { hostId = "exec-1" }); + Assert.Equal(HttpStatusCode.OK, registered.StatusCode); + var reported = await client.PostAsJsonAsync( + "/executors/exec-1/quota-reports", + new { pool = "exec-pool", availablePct = 55.0, observedAt = DateTimeOffset.UtcNow }); + + Assert.Equal(HttpStatusCode.Forbidden, reported.StatusCode); + } + + [Fact] + public async Task MissingBearer_IsUnauthorized() + { + using var client = ClientWith(null); + + var reported = await client.PostAsJsonAsync( + "/executors/exec-1/quota-reports", + new { pool = "exec-pool", availablePct = 55.0, observedAt = DateTimeOffset.UtcNow }); + + Assert.Equal(HttpStatusCode.Unauthorized, reported.StatusCode); + } +} + +internal sealed class QuotaReportAuthFactory : WebApplicationFactory +{ + private readonly string _dbPath = Path.Combine( + Path.GetTempPath(), $"codeybox-quota-auth-{Guid.NewGuid():N}.db"); + + protected override void ConfigureWebHost(IWebHostBuilder builder) + { + builder.UseEnvironment("Development"); + builder.ConfigureAppConfiguration((_, cfg) => + { + var tmp = Path.GetTempPath(); + cfg.AddInMemoryCollection(new Dictionary + { + ["CodeyBox:StateDatabasePath"] = _dbPath, + ["CodeyBox:GitRootDirectory"] = Path.Combine(tmp, $"test-git-{Guid.NewGuid():N}"), + ["CodeyBox:AuditLog:Path"] = Path.Combine(tmp, $"test-log-{Guid.NewGuid():N}-.json"), + ["CodeyBox:AuditLog:AuditPath"] = Path.Combine(tmp, $"test-audit-{Guid.NewGuid():N}-.json"), + ["CodeyBox:QuotaRouter:Pools:exec-pool:Kind"] = "ResettingWindow", + ["CodeyBox:QuotaRouter:Pools:exec-pool:ProbeSource"] = "ExecutorReported", + ["CodeyBox:QuotaRouter:Pools:exec-pool:ReportedReadingMaxAgeSeconds"] = "300", + ["CodeyBox:QuotaRouter:Pools:exec-pool:HolderHostIds:0"] = "exec-1", + }); + }); + builder.ConfigureTestServices(services => + { + services.RemoveAll(); + services.RemoveAll(); + services.AddSingleton(new ApiKeyState(Token: null, Disabled: false, Clients: + [ + new ResolvedApiClient( + "exec-1-client", + "test-bearer-bound-to-exec-1", + new WorkInitiator { Issuer = "test", Subject = "exec-1", DisplayName = "exec-1" }, + CanDelegateInitiator: false, + ExecutorHostId: "exec-1"), + new ResolvedApiClient( + "unbound-client", + "test-bearer-with-no-host-binding", + new WorkInitiator { Issuer = "test", Subject = "service", DisplayName = "service" }, + CanDelegateInitiator: false, + ExecutorHostId: null), + ])); + }); + } + + protected override void Dispose(bool disposing) + { + if (disposing) + { + try { File.Delete(_dbPath); } catch { } + } + base.Dispose(disposing); + } +} diff --git a/tests/CodeyBox.Tests/ExecutorQuotaReportTests.cs b/tests/CodeyBox.Tests/ExecutorQuotaReportTests.cs new file mode 100644 index 00000000..2e8e7237 --- /dev/null +++ b/tests/CodeyBox.Tests/ExecutorQuotaReportTests.cs @@ -0,0 +1,761 @@ +using CodeyBox.Api; +using CodeyBox.Core; +using CodeyBox.Orchestrator; +using Microsoft.Extensions.Logging.Abstractions; + +namespace CodeyBox.Tests; + +/// +/// Verification for executor-reported quota: pools whose credential lives on +/// an executor host are metered from that host's reported readings instead of +/// a direct orchestrator probe. Covers the six acceptance behaviours: +/// orchestrator-held pools behave exactly as today; executor-held pools gate +/// identically to the same reading obtained directly; reports from +/// non-holders are rejected without mutating the stored reading; silence +/// becomes unknown within the staleness bound (fail-closed on a non-zero +/// floor); out-of-range or reset-inconsistent readings are rejected without +/// being stored; and admission is decided by the orchestrator in both modes. +/// +public sealed class ExecutorQuotaReportTests +{ + private static readonly AgentKind Claude = AgentKind.Claude; + private static readonly DateTimeOffset T0 = new(2026, 7, 3, 12, 0, 0, TimeSpan.Zero); + + private sealed class AdvancingClock(DateTimeOffset start) : TimeProvider + { + private DateTimeOffset _now = start; + public override DateTimeOffset GetUtcNow() => _now; + public void Advance(TimeSpan delta) => _now += delta; + } + + private sealed class MutableProbe(AgentKind kind, AgentQuotaSnapshot snapshot) : IAgentQuotaProbe + { + public AgentKind Kind { get; } = kind; + public AgentQuotaSnapshot Current { get; set; } = snapshot; + public int CallCount { get; private set; } + + public Task GetAvailabilityAsync(AgentMembership member, CancellationToken ct) + { + CallCount++; + return Task.FromResult(Current); + } + } + + private static AgentMembership Sub(AgentKind agent, string? pool, int score = 100) => new() + { + Agent = agent, + Billing = AgentBilling.Subscription, + QualityScore = score, + Pool = pool, + }; + + private static AgentClass SoloClass(string id, AgentMembership member) => new() + { + Id = id, + DisplayName = id, + Members = [member], + }; + + private static WorkItem MakeItem(string? classId = null) => new() + { + Id = WorkItemId.New(), + ProjectId = new ProjectId("proj"), + Title = "t", + Prompt = "p", + AgentClassId = classId, + }; + + private static QuotaRouterOptions BaseOpts() => new() + { + MinQuotaPct = 10.0, + StartFloorPct = 25.0, + EndFloorPct = 3.0, + RampWindow = TimeSpan.FromDays(7), + QuotaRecheckInterval = TimeSpan.FromMinutes(5), + UnknownPolicy = QuotaUnknownPolicy.UseObservedFailures, + DispatchReservationEstimatePct = 5.0, + DispatchReservationMinPct = 0.5, + DispatchReservationMaxPct = 25.0, + }; + + private static void ResettingPool( + QuotaRouterOptions opts, + string name, + double floor, + QuotaProbeSource source = QuotaProbeSource.OrchestratorDirect, + string[]? holders = null, + TimeSpan? maxAge = null) + { + opts.Pools[name] = new QuotaPoolOptions + { + Name = name, + Kind = QuotaPoolKind.ResettingWindow, + ProbeSource = source, + ReportedReadingMaxAge = maxAge ?? QuotaRouterDefaults.DefaultReportedReadingMaxAge, + HolderHostIds = holders is null ? [] : [.. holders], + }; + opts.FloorByPool[name] = new QuotaPoolFloorOptions + { + MinQuotaPct = floor, + StartFloorPct = floor, + EndFloorPct = floor, + }; + } + + private static void BalancePool( + QuotaRouterOptions opts, + string name, + double floor, + QuotaProbeSource source = QuotaProbeSource.OrchestratorDirect, + string[]? holders = null) + { + opts.Pools[name] = new QuotaPoolOptions + { + Name = name, + Kind = QuotaPoolKind.DepletingBalance, + BalanceUnit = "credits", + ProbeSource = source, + HolderHostIds = holders is null ? [] : [.. holders], + }; + opts.FloorByPool[name] = new QuotaPoolFloorOptions { MinBalance = floor }; + } + + private static AgentClassRouter BuildRouter( + AgentClass catalog, + IEnumerable probes, + QuotaRouterOptions opts, + AdvancingClock clock, + ExecutorQuotaReportStore? store) => + new( + [catalog], + probes, + opts, + NullLogger.Instance, + timeProvider: clock, + reportStore: store); + + private static ExecutorQuotaReport PctReport(string pool, double pct, DateTimeOffset observed) => new() + { + PoolName = pool, + AvailablePct = pct, + ObservedAt = observed, + }; + + // ── 1. Orchestrator-held pools behave exactly as today ─────────────────── + + [Fact] + public async Task OrchestratorDirectPool_ProbesDirectlyAndRejectsExecutorReports() + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + ResettingPool(opts, "local", floor: 20); + var probe = new MutableProbe(Claude, new AgentQuotaSnapshot { AvailablePct = 50 }); + var store = new ExecutorQuotaReportStore(opts, clock); + var router = BuildRouter(SoloClass("c", Sub(Claude, "local")), [probe], opts, clock, store); + + var decision = await router.ResolveAsync(MakeItem("c"), null, CancellationToken.None); + + Assert.NotNull(decision.Chosen); + Assert.Equal(1, probe.CallCount); + + Assert.False(store.TryReport("exec-1", PctReport("local", 90, clock.GetUtcNow()), out var reason)); + Assert.Contains("orchestrator-probed", reason, StringComparison.OrdinalIgnoreCase); + Assert.False(store.TryGetStored("local", out _, out _)); + + var again = await router.ResolveAsync(MakeItem("c"), null, CancellationToken.None); + Assert.NotNull(again.Chosen); + Assert.Equal(2, probe.CallCount); + } + + // ── 2. Executor-held pools gate identically to the same direct reading ─── + + [Fact] + public async Task ExecutorReportedPool_GateMatchesDirectProbeForSameReading() + { + foreach (var pct in new[] { 80.0, 10.0 }) + { + var clock = new AdvancingClock(T0); + + var directOpts = BaseOpts(); + ResettingPool(directOpts, "pool", floor: 20); + var directProbe = new MutableProbe(Claude, new AgentQuotaSnapshot { AvailablePct = pct }); + var directRouter = BuildRouter( + SoloClass("c", Sub(Claude, "pool")), [directProbe], directOpts, clock, null); + var direct = await directRouter.ResolveAsync(MakeItem("c"), null, CancellationToken.None); + + var execOpts = BaseOpts(); + ResettingPool(execOpts, "pool", floor: 20, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var execProbe = new MutableProbe(Claude, new AgentQuotaSnapshot { AvailablePct = -1 }); + var execStore = new ExecutorQuotaReportStore(execOpts, clock); + Assert.True(execStore.TryReport("exec-1", PctReport("pool", pct, clock.GetUtcNow()), out _)); + var execRouter = BuildRouter( + SoloClass("c", Sub(Claude, "pool")), [execProbe], execOpts, clock, execStore); + var reported = await execRouter.ResolveAsync(MakeItem("c"), null, CancellationToken.None); + + Assert.Equal(direct.Chosen?.RouteKey, reported.Chosen?.RouteKey); + Assert.Equal(direct.ShouldWait, reported.ShouldWait); + Assert.Equal(0, execProbe.CallCount); + if (pct >= 20) + Assert.NotNull(reported.Chosen); + else + { + Assert.Null(reported.Chosen); + Assert.True(reported.ShouldWait); + } + } + } + + [Fact] + public async Task ExecutorReportedBalancePool_GateMatchesDirectProbeForSameBalance() + { + foreach (var (balance, expectChosen) in new[] { (1000.0, true), (100.0, false) }) + { + var clock = new AdvancingClock(T0); + + var directOpts = BaseOpts(); + BalancePool(directOpts, "prepaid", floor: 500); + var directProbe = new MutableProbe(Claude, + new AgentQuotaSnapshot { AvailablePct = -1, BalanceRemaining = balance, BalanceUnit = "credits" }); + var directRouter = BuildRouter( + SoloClass("c", Sub(Claude, "prepaid")), [directProbe], directOpts, clock, null); + var direct = await directRouter.ResolveAsync(MakeItem("c"), null, CancellationToken.None); + + var execOpts = BaseOpts(); + BalancePool(execOpts, "prepaid", floor: 500, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var execProbe = new MutableProbe(Claude, + AgentQuotaSnapshot.UnknownSnapshot(QuotaUnknownReason.NoCredential, "no local credential")); + var execStore = new ExecutorQuotaReportStore(execOpts, clock); + Assert.True(execStore.TryReport("exec-1", + new ExecutorQuotaReport + { + PoolName = "prepaid", + BalanceRemaining = balance, + ObservedAt = clock.GetUtcNow(), + }, out _)); + var execRouter = BuildRouter( + SoloClass("c", Sub(Claude, "prepaid")), [execProbe], execOpts, clock, execStore); + var reported = await execRouter.ResolveAsync(MakeItem("c"), null, CancellationToken.None); + + Assert.Equal(direct.Chosen?.RouteKey, reported.Chosen?.RouteKey); + Assert.Equal(direct.ShouldWait, reported.ShouldWait); + Assert.Equal(expectChosen, reported.Chosen is not null); + Assert.Equal(0, execProbe.CallCount); + } + } + + // ── 3. Non-holder reports are rejected; the stored reading is unchanged ── + + [Fact] + public void ReportFromUndeclaredHost_IsRejectedAndStoredReadingUnchanged() + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + ResettingPool(opts, "pool", floor: 20, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var store = new ExecutorQuotaReportStore(opts, clock); + Assert.True(store.TryReport("exec-1", PctReport("pool", 80, clock.GetUtcNow()), out _)); + + foreach (var impostor in new[] { "exec-2", "exec-1x", "EXEC-1" }) + { + Assert.False(store.TryReport(impostor, PctReport("pool", 5, clock.GetUtcNow()), out var reason)); + Assert.Contains("not declared as holding", reason, StringComparison.OrdinalIgnoreCase); + } + + foreach (var empty in new[] { "", " " }) + { + Assert.False(store.TryReport(empty, PctReport("pool", 5, clock.GetUtcNow()), out var reason)); + Assert.Contains("host id is required", reason, StringComparison.OrdinalIgnoreCase); + } + + Assert.True(store.TryGetStored("pool", out var stored, out var holder)); + Assert.NotNull(stored); + Assert.Equal(80, stored!.AvailablePct); + Assert.Equal("exec-1", holder); + Assert.Equal(80, store.GetSnapshot("pool").AvailablePct); + } + + // ── 4. Silence becomes unknown within the staleness bound ──────────────── + + [Fact] + public async Task SilentExecutor_BecomesUnknownWithinStalenessBound_AndFailClosed() + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + ResettingPool(opts, "pool", floor: 20, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"], + maxAge: TimeSpan.FromMinutes(5)); + var probe = new MutableProbe(Claude, + AgentQuotaSnapshot.UnknownSnapshot(QuotaUnknownReason.NoCredential, "no local credential")); + var store = new ExecutorQuotaReportStore(opts, clock); + var router = BuildRouter(SoloClass("c", Sub(Claude, "pool")), [probe], opts, clock, store); + + Assert.True(store.TryReport("exec-1", PctReport("pool", 80, clock.GetUtcNow()), out _)); + var healthy = await router.ResolveAsync(MakeItem("c"), null, CancellationToken.None); + Assert.NotNull(healthy.Chosen); + + clock.Advance(TimeSpan.FromMinutes(5).Add(TimeSpan.FromSeconds(1))); + var stale = store.GetSnapshot("pool"); + Assert.False(stale.IsKnown); + Assert.Equal(QuotaUnknownReason.Transient, stale.Unknown); + + var denied = await router.ResolveAsync(MakeItem("c"), null, CancellationToken.None); + Assert.Null(denied.Chosen); + Assert.True(denied.ShouldWait); + Assert.Contains("floor", denied.Reason, StringComparison.OrdinalIgnoreCase); + Assert.Equal(0, probe.CallCount); + } + + [Fact] + public async Task NeverReportedExecutorPool_ReadsUnknownAndFailClosedOnFloor() + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + ResettingPool(opts, "pool", floor: 20, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var probe = new MutableProbe(Claude, + AgentQuotaSnapshot.UnknownSnapshot(QuotaUnknownReason.NoCredential, "no local credential")); + var store = new ExecutorQuotaReportStore(opts, clock); + var router = BuildRouter(SoloClass("c", Sub(Claude, "pool")), [probe], opts, clock, store); + + var missing = store.GetSnapshot("pool"); + Assert.False(missing.IsKnown); + Assert.Equal(QuotaUnknownReason.Transient, missing.Unknown); + + var denied = await router.ResolveAsync(MakeItem("c"), null, CancellationToken.None); + Assert.Null(denied.Chosen); + Assert.True(denied.ShouldWait); + } + + [Fact] + public void FreshExecutorUnknown_PreservesReasonLikeDirectProbe() + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + ResettingPool(opts, "pool", floor: 20, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var store = new ExecutorQuotaReportStore(opts, clock); + + foreach (var reason in new[] { QuotaUnknownReason.Transient, QuotaUnknownReason.Permanent, QuotaUnknownReason.NoCredential }) + { + Assert.True(store.TryReport("exec-1", + new ExecutorQuotaReport + { + PoolName = "pool", + ObservedAt = clock.GetUtcNow(), + Unknown = reason, + }, out _)); + var snapshot = store.GetSnapshot("pool"); + Assert.False(snapshot.IsKnown); + Assert.Equal(reason, snapshot.Unknown); + } + } + + // ── 5. Out-of-range / reset-inconsistent readings are rejected ─────────── + + [Theory] + [InlineData(101.0)] + [InlineData(-0.5)] + public void OutOfRangePercentage_IsRejectedWithoutStoring(double pct) + { + RejectPct(pct); + } + + [Fact] + public void NonFinitePercentage_IsRejectedWithoutStoring() + { + RejectPct(double.NaN); + RejectPct(double.PositiveInfinity); + RejectPct(double.NegativeInfinity); + } + + private static void RejectPct(double pct) + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + ResettingPool(opts, "pool", floor: 20, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var store = new ExecutorQuotaReportStore(opts, clock); + Assert.True(store.TryReport("exec-1", PctReport("pool", 60, clock.GetUtcNow()), out _)); + + Assert.False(store.TryReport("exec-1", + new ExecutorQuotaReport { PoolName = "pool", AvailablePct = pct, ObservedAt = clock.GetUtcNow() }, + out var reason)); + Assert.Contains("0-100", reason, StringComparison.Ordinal); + + Assert.True(store.TryGetStored("pool", out var stored, out _)); + Assert.Equal(60, stored!.AvailablePct); + } + + [Fact] + public void MissingPercentageOnResettingPool_IsRejectedWithoutStoring() + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + ResettingPool(opts, "pool", floor: 20, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var store = new ExecutorQuotaReportStore(opts, clock); + Assert.True(store.TryReport("exec-1", PctReport("pool", 60, clock.GetUtcNow()), out _)); + + Assert.False(store.TryReport("exec-1", + new ExecutorQuotaReport { PoolName = "pool", ObservedAt = clock.GetUtcNow() }, + out _)); + + Assert.True(store.TryGetStored("pool", out var stored, out _)); + Assert.Equal(60, stored!.AvailablePct); + } + + [Fact] + public void InvalidBalance_IsRejectedWithoutStoring() + { + RejectBalance(-1.0); + RejectBalance(double.NaN); + RejectBalance(double.PositiveInfinity); + } + + private static void RejectBalance(double balance) + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + BalancePool(opts, "prepaid", floor: 100, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var store = new ExecutorQuotaReportStore(opts, clock); + Assert.True(store.TryReport("exec-1", + new ExecutorQuotaReport + { + PoolName = "prepaid", + BalanceRemaining = 1000, + ObservedAt = clock.GetUtcNow(), + }, out _)); + + Assert.False(store.TryReport("exec-1", + new ExecutorQuotaReport + { + PoolName = "prepaid", + BalanceRemaining = balance, + ObservedAt = clock.GetUtcNow(), + }, out var reason)); + Assert.Contains("balance", reason, StringComparison.OrdinalIgnoreCase); + + Assert.True(store.TryGetStored("prepaid", out var stored, out _)); + Assert.Equal(1000, stored!.BalanceRemaining); + } + + [Fact] + public void ResetOnBalancePool_IsRejectedWithoutStoring() + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + BalancePool(opts, "prepaid", floor: 100, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var store = new ExecutorQuotaReportStore(opts, clock); + Assert.True(store.TryReport("exec-1", + new ExecutorQuotaReport + { + PoolName = "prepaid", + BalanceRemaining = 1000, + ObservedAt = clock.GetUtcNow(), + }, out _)); + + Assert.False(store.TryReport("exec-1", + new ExecutorQuotaReport + { + PoolName = "prepaid", + BalanceRemaining = 900, + ResetAt = clock.GetUtcNow().AddDays(7), + ObservedAt = clock.GetUtcNow(), + }, out var reason)); + Assert.Contains("reset", reason, StringComparison.OrdinalIgnoreCase); + + Assert.True(store.TryGetStored("prepaid", out var stored, out _)); + Assert.Equal(1000, stored!.BalanceRemaining); + Assert.Null(store.GetSnapshot("prepaid").ResetAt); + } + + [Fact] + public void ResetOnResettingPool_IsAccepted() + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + ResettingPool(opts, "pool", floor: 20, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var store = new ExecutorQuotaReportStore(opts, clock); + var reset = clock.GetUtcNow().AddDays(7); + + Assert.True(store.TryReport("exec-1", + new ExecutorQuotaReport + { + PoolName = "pool", + AvailablePct = 60, + ResetAt = reset, + ObservedAt = clock.GetUtcNow(), + }, out _)); + Assert.Equal(reset, store.GetSnapshot("pool").ResetAt); + } + + [Fact] + public void FutureObservedBeyondSkew_AndUnknownPool_AreRejected() + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + ResettingPool(opts, "pool", floor: 20, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var store = new ExecutorQuotaReportStore(opts, clock); + + Assert.False(store.TryReport("exec-1", + PctReport("pool", 60, clock.GetUtcNow().AddHours(1)), out var futureReason)); + Assert.Contains("future", futureReason, StringComparison.OrdinalIgnoreCase); + + Assert.False(store.TryReport("exec-1", + PctReport("ghost", 60, clock.GetUtcNow()), out var ghostReason)); + Assert.Contains("no configured quota pool", ghostReason, StringComparison.OrdinalIgnoreCase); + + Assert.False(store.TryGetStored("pool", out _, out _)); + } + + // ── 6. Admission is decided by the orchestrator in both modes ──────────── + + [Fact] + public async Task HealthyExecutorReport_DoesNotAdmitPastOrchestratorFloor() + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + ResettingPool(opts, "pool", floor: 20, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var probe = new MutableProbe(Claude, + AgentQuotaSnapshot.UnknownSnapshot(QuotaUnknownReason.NoCredential, "no local credential")); + var store = new ExecutorQuotaReportStore(opts, clock); + Assert.True(store.TryReport("exec-1", PctReport("pool", 80, clock.GetUtcNow()), out _)); + var router = BuildRouter(SoloClass("c", Sub(Claude, "pool")), [probe], opts, clock, store); + + var admitted = await router.ResolveAsync(MakeItem("c"), null, CancellationToken.None); + Assert.NotNull(admitted.Chosen); + + opts.FloorByPool["pool"] = new QuotaPoolFloorOptions + { + MinQuotaPct = 90, + StartFloorPct = 90, + EndFloorPct = 90, + }; + var denied = await router.ResolveAsync(MakeItem("c"), null, CancellationToken.None); + Assert.Null(denied.Chosen); + Assert.True(denied.ShouldWait); + Assert.Equal(80, store.GetSnapshot("pool").AvailablePct); + } + + [Fact] + public void ReportStore_ExposesNoAdmissionDecision() + { + var admissionLike = typeof(ExecutorQuotaReportStore) + .GetMethods(System.Reflection.BindingFlags.Public | System.Reflection.BindingFlags.Instance) + .Where(m => m.Name.Contains("Allow", StringComparison.OrdinalIgnoreCase) + || m.Name.Contains("Admit", StringComparison.OrdinalIgnoreCase) + || m.Name.Contains("Gate", StringComparison.OrdinalIgnoreCase) + || m.Name.Contains("Decide", StringComparison.OrdinalIgnoreCase) + || m.ReturnType == typeof(QuotaGateDecision)) + .Select(m => m.Name) + .ToArray(); + Assert.Empty(admissionLike); + } + + // ── Configuration mapping ──────────────────────────────────────────────── + + [Fact] + public void PoolConfig_MapsProbeSourceHoldersAndStaleness() + { + var config = new QuotaRouterConfig + { + Pools = + { + ["remote"] = new QuotaPoolConfig + { + Kind = "ResettingWindow", + ProbeSource = "ExecutorReported", + ReportedReadingMaxAgeSeconds = 120, + HolderHostIds = ["exec-1", "exec-2"], + }, + ["local"] = new QuotaPoolConfig { Kind = "ResettingWindow" }, + }, + }; + + var mapped = QuotaRouterConfigMapper.ToOptions(config); + + Assert.Equal(QuotaProbeSource.ExecutorReported, mapped.Pools["remote"].ProbeSource); + Assert.Equal(TimeSpan.FromSeconds(120), mapped.Pools["remote"].ReportedReadingMaxAge); + Assert.Equal(new[] { "exec-1", "exec-2" }, mapped.Pools["remote"].HolderHostIds); + Assert.Equal(QuotaProbeSource.OrchestratorDirect, mapped.Pools["local"].ProbeSource); + Assert.Equal( + QuotaRouterDefaults.DefaultReportedReadingMaxAge, + mapped.Pools["local"].ReportedReadingMaxAge); + Assert.Empty(mapped.Pools["local"].HolderHostIds); + } + + [Theory] + [InlineData("executorreported")] + [InlineData("EXECUTORREPORTED")] + [InlineData(" ExecutorReported ")] + public void PoolConfig_ProbeSourceParsesCaseInsensitively(string source) + { + var config = new QuotaRouterConfig + { + Pools = { ["p"] = new QuotaPoolConfig { Kind = "ResettingWindow", ProbeSource = source } }, + }; + Assert.Equal(QuotaProbeSource.ExecutorReported, QuotaRouterConfigMapper.ToOptions(config).Pools["p"].ProbeSource); + } + + [Fact] + public void PoolConfig_RejectsUnknownProbeSourceNegativeStalenessAndEmptyHolder() + { + Assert.Throws(() => QuotaRouterConfigMapper.ToOptions(new QuotaRouterConfig + { + Pools = { ["p"] = new QuotaPoolConfig { Kind = "ResettingWindow", ProbeSource = "SomewhereElse" } }, + })); + Assert.Throws(() => QuotaRouterConfigMapper.ToOptions(new QuotaRouterConfig + { + Pools = + { + ["p"] = new QuotaPoolConfig + { + Kind = "ResettingWindow", + ProbeSource = "ExecutorReported", + ReportedReadingMaxAgeSeconds = -5, + }, + }, + })); + Assert.Throws(() => QuotaRouterConfigMapper.ToOptions(new QuotaRouterConfig + { + Pools = + { + ["p"] = new QuotaPoolConfig + { + Kind = "ResettingWindow", + ProbeSource = "ExecutorReported", + HolderHostIds = ["exec-1", ""], + }, + }, + })); + } + + // ── Bounded resets, bounded notes, kind-change safety ──────────────────── + + [Fact] + public void FarFutureReset_IsRejectedWithoutStoring() + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + ResettingPool(opts, "pool", floor: 20, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var store = new ExecutorQuotaReportStore(opts, clock); + Assert.True(store.TryReport("exec-1", PctReport("pool", 60, clock.GetUtcNow()), out _)); + + Assert.False(store.TryReport("exec-1", + new ExecutorQuotaReport + { + PoolName = "pool", + AvailablePct = 60, + ResetAt = clock.GetUtcNow().AddDays(30), + ObservedAt = clock.GetUtcNow(), + }, out var reason)); + Assert.Contains("horizon", reason, StringComparison.OrdinalIgnoreCase); + + Assert.True(store.TryGetStored("pool", out var stored, out _)); + Assert.Equal(60, stored!.AvailablePct); + Assert.Null(stored.ResetAt); + } + + [Fact] + public void LongPastReset_IsRejectedWithoutStoring() + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + ResettingPool(opts, "pool", floor: 20, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var store = new ExecutorQuotaReportStore(opts, clock); + Assert.True(store.TryReport("exec-1", PctReport("pool", 60, clock.GetUtcNow()), out _)); + + Assert.False(store.TryReport("exec-1", + new ExecutorQuotaReport + { + PoolName = "pool", + AvailablePct = 60, + ResetAt = clock.GetUtcNow().AddDays(-2), + ObservedAt = clock.GetUtcNow(), + }, out var reason)); + Assert.Contains("observed", reason, StringComparison.OrdinalIgnoreCase); + + Assert.True(store.TryGetStored("pool", out var stored, out _)); + Assert.Equal(60, stored!.AvailablePct); + } + + [Fact] + public void OversizedNotes_AreRejectedWithoutStoring() + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + ResettingPool(opts, "pool", floor: 20, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var store = new ExecutorQuotaReportStore(opts, clock); + Assert.True(store.TryReport("exec-1", PctReport("pool", 60, clock.GetUtcNow()), out _)); + + Assert.False(store.TryReport("exec-1", + new ExecutorQuotaReport + { + PoolName = "pool", + AvailablePct = 60, + ObservedAt = clock.GetUtcNow(), + Notes = new string('n', ExecutorQuotaReportStore.MaxReportNotesLength + 1), + }, out var reason)); + Assert.Contains("notes", reason, StringComparison.OrdinalIgnoreCase); + + Assert.True(store.TryGetStored("pool", out var stored, out _)); + Assert.Equal(60, stored!.AvailablePct); + } + + [Fact] + public void NotesWithControlCharacters_AreRejectedWithoutStoring() + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + ResettingPool(opts, "pool", floor: 20, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var store = new ExecutorQuotaReportStore(opts, clock); + Assert.True(store.TryReport("exec-1", PctReport("pool", 60, clock.GetUtcNow()), out _)); + + Assert.False(store.TryReport("exec-1", + new ExecutorQuotaReport + { + PoolName = "pool", + AvailablePct = 60, + ObservedAt = clock.GetUtcNow(), + Notes = "probe ok\u0000injected", + }, out var reason)); + Assert.Contains("control", reason, StringComparison.OrdinalIgnoreCase); + + Assert.True(store.TryGetStored("pool", out var stored, out _)); + Assert.Equal(60, stored!.AvailablePct); + } + + [Fact] + public void BalanceReportFollowedByKindHotReload_ReadsTransientUnknownInsteadOfThrowing() + { + var clock = new AdvancingClock(T0); + var opts = BaseOpts(); + BalancePool(opts, "prepaid", floor: 100, + source: QuotaProbeSource.ExecutorReported, holders: ["exec-1"]); + var store = new ExecutorQuotaReportStore(opts, clock); + Assert.True(store.TryReport("exec-1", + new ExecutorQuotaReport + { + PoolName = "prepaid", + BalanceRemaining = 1000, + ObservedAt = clock.GetUtcNow(), + }, out _)); + + opts.Pools["prepaid"].Kind = QuotaPoolKind.ResettingWindow; + + var snapshot = store.GetSnapshot("prepaid"); + Assert.False(snapshot.IsKnown); + Assert.Equal(QuotaUnknownReason.Transient, snapshot.Unknown); + } +}