Add WorkerProxy assignment, readiness, and state polling APIs

src/Functions.WorkerProxy/Http/WorkerHttpCapabilityProvider.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.Collections.Generic;
+using Azure.Functions.WorkerProxy.Rpc;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 
@@ -12,6 +13,7 @@ namespace Azure.Functions.WorkerProxy.Http;
 /// Captures the worker's HTTP destination before advertising WorkerProxy to the runtime.
 /// </summary>
 internal sealed partial class WorkerHttpCapabilityProvider(IOptions<WorkerProxyOptions> options, ILogger<WorkerHttpCapabilityProvider> logger)
+    : IWorkerCapabilityFinalizer
 {
     private const string HttpUriCapability = "HttpUri";
 

src/Functions.WorkerProxy/Management/ManagementApiEndpoints.cs

@@ -0,0 +1,81 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System;
+using System.Globalization;
+using System.Text.Json;
+using System.Threading.Tasks;
+using Azure.Functions.WorkerProxy.State;
+using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Http;
+using Microsoft.AspNetCore.Routing;
+
+namespace Azure.Functions.WorkerProxy.Management;
+
+/// <summary>
+/// Registers worker lifecycle APIs on the management listener.
+/// </summary>
+/// <remarks>
+/// The PUT handler explicitly uses ReadFromJsonAsync rather than automatic body binding so malformed JSON,
+/// incompatible field types, and unsupported content types return our Host-aligned HTTP 400 InvalidBody
+/// validation envelope. Automatic binding can reject requests before the handler runs with framework-owned
+/// 400/415 responses that do not guarantee that envelope.
+/// </remarks>
+internal static class ManagementApiEndpoints
+{
+    public static void Map(IEndpointRouteBuilder endpoints)
+    {
+        endpoints.MapGet("/admin/worker/ready", ManagementApiHandlers.GetWorkerReady)
+            .AddEndpointFilter(DisableCaching).AllowAnonymous();
+        endpoints.MapPut("/admin/worker/assignment", AssignWorkerAsync).AllowAnonymous();
+        endpoints.MapGet("/admin/worker/state", GetInstanceStateAsync)
+            .AddEndpointFilter(DisableCaching).AllowAnonymous();
+    }
+
+    private static ValueTask<object?> DisableCaching(EndpointFilterInvocationContext context, EndpointFilterDelegate next)
+    {
+        context.HttpContext.Response.Headers.CacheControl = "no-store";
+        return next(context);
+    }
+
+    private static async Task<IResult> AssignWorkerAsync(HttpRequest request, WorkerPodStateManager manager)
+    {
+        if (!request.HasJsonContentType())
+        {
+            return ManagementApiHandlers.InvalidBody();
+        }
+
+        WorkerAssignRequest? assignment;
+        try
+        {
+            assignment = await request.ReadFromJsonAsync(
+                WorkerProxyJsonContext.Default.WorkerAssignRequest, request.HttpContext.RequestAborted);
+        }
+        catch (JsonException)
+        {
+            return ManagementApiHandlers.InvalidBody();
+        }
+
+        return ManagementApiHandlers.AssignWorker(assignment, manager);
+    }
+
+    private static async Task<IResult> GetInstanceStateAsync(HttpRequest request, WorkerPodStateManager manager)
+    {
+        long? lastKnownRevision = null;
+        if (request.Query.TryGetValue("lastKnownRevision", out var revisions))
+        {
+            string? value = revisions.Count == 1 ? revisions[0] : null;
+            // Accept invariant ASCII digits with an optional leading sign, not whitespace or JSON syntax.
+            if (value is null
+                || value.AsSpan().IndexOfAnyExcept("+-0123456789") >= 0
+                || !long.TryParse(value, NumberStyles.AllowLeadingSign, CultureInfo.InvariantCulture, out long revision))
+            {
+                return ManagementApiHandlers.InvalidRevision();
+            }
+
+            lastKnownRevision = revision;
+        }
+
+        return await ManagementApiHandlers.GetInstanceStateAsync(lastKnownRevision, manager, request.HttpContext.RequestAborted);
+    }
+}

src/Functions.WorkerProxy/Management/ManagementApiHandlers.cs

@@ -0,0 +1,135 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure.Functions.WorkerProxy.State;
+using Microsoft.AspNetCore.Http;
+
+namespace Azure.Functions.WorkerProxy.Management;
+
+/// <summary>
+/// Maps validated management requests and worker lifecycle outcomes to HTTP results.
+/// </summary>
+internal static class ManagementApiHandlers
+{
+    public static IResult GetWorkerReady(WorkerPodStateManager manager) =>
+        manager.State.IsWorkerReady ? TypedResults.Ok() : TypedResults.StatusCode(StatusCodes.Status503ServiceUnavailable);
+
+    public static IResult AssignWorker(WorkerAssignRequest? request, WorkerPodStateManager manager)
+    {
+        if (request is null)
+        {
+            return InvalidBody();
+        }
+
+        List<RequestValidationError> errors = [];
+        if (string.IsNullOrWhiteSpace(request.FunctionAppName))
+        {
+            errors.Add(new(WorkerApiErrorCodes.Required, "functionAppName"));
+        }
+
+        if (string.IsNullOrWhiteSpace(request.FunctionGroupName))
+        {
+            errors.Add(new(WorkerApiErrorCodes.Required, "functionGroupName"));
+        }
+
+        if (request.IsAlwaysReady is null)
+        {
+            errors.Add(new(WorkerApiErrorCodes.Required, "isAlwaysReady"));
+        }
+
+        if (string.IsNullOrWhiteSpace(request.FunctionAppDirectory))
+        {
+            errors.Add(new(WorkerApiErrorCodes.Required, "functionAppDirectory"));
+        }
+
+        Dictionary<string, string> environment = new(StringComparer.Ordinal);
+        if (request.Environment is null)
+        {
+            errors.Add(new(WorkerApiErrorCodes.Required, "environment"));
+        }
+        else
+        {
+            foreach ((string key, string? value) in request.Environment)
+            {
+                if (string.IsNullOrEmpty(key) || value is null)
+                {
+                    // Report the invalid field once without exposing environment keys or values.
+                    errors.Add(new(WorkerApiErrorCodes.InvalidValue, "environment"));
+                    break;
+                }
+
+                environment.Add(key, value);
+            }
+        }
+
+        if (errors.Count > 0
+            || request.FunctionAppName is not { } functionAppName
+            || request.FunctionGroupName is not { } functionGroupName
+            || request.FunctionAppDirectory is not { } functionAppDirectory
+            || request.IsAlwaysReady is not { } isAlwaysReady)
+        {
+            return ValidationError(errors);
+        }
+
+        WorkerAssignment assignment = new(
+            functionAppName,
+            functionGroupName,
+            isAlwaysReady,
+            environment,
+            functionAppDirectory);
+        return manager.Assign(assignment) switch
+        {
+            WorkerAssignmentResult.Created => TypedResults.Created("/admin/worker/assignment"),
+            WorkerAssignmentResult.AlreadyAssigned => TypedResults.NoContent(),
+            WorkerAssignmentResult.WorkerNotReady => Error(
+                StatusCodes.Status503ServiceUnavailable, WorkerApiErrorCodes.WorkerNotReady, "The worker has not established a valid StartStream."),
+            WorkerAssignmentResult.AssignmentConflict => Error(
+                StatusCodes.Status409Conflict, WorkerApiErrorCodes.AssignmentConflict, "The pod is already assigned to a different assignment."),
+            WorkerAssignmentResult.WorkerTerminated => Error(
+                StatusCodes.Status409Conflict, WorkerApiErrorCodes.WorkerTerminated, "The assigned worker stream has terminated."),
+            _ => throw new InvalidOperationException("Unexpected worker assignment result.")
+        };
+    }
+
+    public static async Task<IResult> GetInstanceStateAsync(
+        long? revision,
+        WorkerPodStateManager manager,
+        CancellationToken cancellationToken = default)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        if (revision is not { } lastKnownRevision)
+        {
+            return StateResponse(manager.State);
+        }
+
+        // Revisions never decrease, so a revision valid here remains valid when the manager registers the poll.
+        if (lastKnownRevision < 0 || lastKnownRevision > manager.State.Revision)
+        {
+            return InvalidRevision();
+        }
+
+        WorkerStatePollResult result = await manager.WaitForChangeAsync(lastKnownRevision, cancellationToken);
+        return result.State is { } state ? StateResponse(state) : TypedResults.NoContent();
+    }
+
+    internal static IResult InvalidBody() =>
+        ValidationError([new(WorkerApiErrorCodes.InvalidBody, "request")]);
+
+    internal static IResult InvalidRevision() =>
+        ValidationError([new(WorkerApiErrorCodes.InvalidRevision, "lastKnownRevision")]);
+
+    private static IResult ValidationError(IReadOnlyList<RequestValidationError> errors) =>
+        TypedResults.Json(new RequestValidationResponse(errors),
+            WorkerProxyJsonContext.Default.RequestValidationResponse, statusCode: StatusCodes.Status400BadRequest);
+
+    private static IResult Error(int statusCode, string code, string detail) =>
+        TypedResults.Json(new WorkerApiErrorResponse(new(code, detail)),
+            WorkerProxyJsonContext.Default.WorkerApiErrorResponse, statusCode: statusCode);
+
+    private static IResult StateResponse(WorkerPodState state) =>
+        TypedResults.Json(WorkerInstanceState.FromState(state), WorkerProxyJsonContext.Default.WorkerInstanceState);
+}

src/Functions.WorkerProxy/Management/RequestValidationError.cs

@@ -0,0 +1,9 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+namespace Azure.Functions.WorkerProxy.Management;
+
+/// <summary>
+/// Identifies an invalid request field, or request for a body-level error, without echoing its value.
+/// </summary>
+internal sealed record RequestValidationError(string Code, string Target);

src/Functions.WorkerProxy/Management/RequestValidationResponse.cs

@@ -0,0 +1,11 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System.Collections.Generic;
+
+namespace Azure.Functions.WorkerProxy.Management;
+
+/// <summary>
+/// Contains detected field errors, or one request-body error, returned with HTTP 400.
+/// </summary>
+internal sealed record RequestValidationResponse(IReadOnlyList<RequestValidationError> Errors);

src/Functions.WorkerProxy/Management/WorkerApiError.cs

@@ -0,0 +1,17 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+namespace Azure.Functions.WorkerProxy.Management;
+
+/// <summary>
+/// Provides a stable error code and optional diagnostic detail without echoing request values.
+/// </summary>
+/// <param name="Code">A case-sensitive contract identifier; existing codes must not be renamed or repurposed.</param>
+/// <param name="Detail">Diagnostic text that may change and must not be used for client decisions.</param>
+/// <remarks>
+/// WorkerNotReady (503) permits retry after readiness. WorkerTerminated (409) is terminal for the
+/// assigned session; retrying the same assignment on this pod cannot recover it.
+/// AssignmentConflict (409) rejects a different assignment; do not retry that request unchanged.
+/// Clients must inspect Code to distinguish the two 409 outcomes and handle unknown codes gracefully.
+/// </remarks>
+internal sealed record WorkerApiError(string Code, string? Detail = null);

src/Functions.WorkerProxy/Management/WorkerApiErrorCodes.cs

@@ -0,0 +1,20 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+namespace Azure.Functions.WorkerProxy.Management;
+
+/// <summary>
+/// Defines the stable error codes returned by the management APIs.
+/// </summary>
+internal static class WorkerApiErrorCodes
+{
+    // Clients branch on these exact, case-sensitive wire values. Do not change or repurpose them.
+    // Keep explicit literals rather than nameof so symbol renames cannot change the contract.
+    public const string Required = "Required";
+    public const string InvalidBody = "InvalidBody";
+    public const string InvalidValue = "InvalidValue";
+    public const string InvalidRevision = "InvalidRevision";
+    public const string WorkerNotReady = "WorkerNotReady";
+    public const string WorkerTerminated = "WorkerTerminated";
+    public const string AssignmentConflict = "AssignmentConflict";
+}

src/Functions.WorkerProxy/Management/WorkerApiErrorResponse.cs

@@ -0,0 +1,9 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+namespace Azure.Functions.WorkerProxy.Management;
+
+/// <summary>
+/// Wraps validation and lifecycle failures in the same management API error envelope.
+/// </summary>
+internal sealed record WorkerApiErrorResponse(WorkerApiError Error);

src/Functions.WorkerProxy/Management/WorkerAssignRequest.cs

@@ -0,0 +1,22 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System.Collections.Generic;
+
+namespace Azure.Functions.WorkerProxy.Management;
+
+/// <summary>
+/// Describes assignment identity for an already-specialized worker.
+/// </summary>
+internal sealed class WorkerAssignRequest
+{
+    public string? FunctionAppName { get; init; }
+
+    public string? FunctionGroupName { get; init; }
+
+    public bool? IsAlwaysReady { get; init; }
+
+    public Dictionary<string, string?>? Environment { get; init; }
+
+    public string? FunctionAppDirectory { get; init; }
+}

src/Functions.WorkerProxy/Management/WorkerInstanceState.cs

@@ -0,0 +1,20 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using Azure.Functions.WorkerProxy.State;
+
+namespace Azure.Functions.WorkerProxy.Management;
+
+/// <summary>
+/// Exposes platform-facing pod state without internal session details or assignment environment values.
+/// </summary>
+internal sealed record WorkerInstanceState(
+    string PodName,
+    long RevisionId,
+    WorkerPodStateResponse WorkerPodState)
+{
+    public string FunctionsContainerType => "FunctionsWorkerPod";
+
+    public static WorkerInstanceState FromState(WorkerPodState state) =>
+        new(state.PodName, state.Revision, new(state.PodStatus, state.FunctionGroupName, state.IsAlwaysReady));
+}

src/Functions.WorkerProxy/Management/WorkerPodStateResponse.cs

@@ -0,0 +1,15 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System.Text.Json.Serialization;
+using Azure.Functions.WorkerProxy.State;
+
+namespace Azure.Functions.WorkerProxy.Management;
+
+/// <summary>
+/// Reports worker-pod eligibility and known assignment identity, not runtime serving readiness.
+/// </summary>
+internal sealed record WorkerPodStateResponse(
+    [property: JsonConverter(typeof(JsonStringEnumConverter<WorkerPodStatus>))] WorkerPodStatus PodStatus,
+    string? FunctionGroupName,
+    bool? IsAlwaysReady);

src/Functions.WorkerProxy/Management/WorkerProxyJsonContext.cs

@@ -0,0 +1,16 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System.Text.Json.Serialization;
+
+namespace Azure.Functions.WorkerProxy.Management;
+
+[JsonSourceGenerationOptions(
+    PropertyNamingPolicy = JsonKnownNamingPolicy.CamelCase,
+    PropertyNameCaseInsensitive = true,
+    DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull)]
+[JsonSerializable(typeof(WorkerAssignRequest))]
+[JsonSerializable(typeof(WorkerInstanceState))]
+[JsonSerializable(typeof(WorkerApiErrorResponse))]
+[JsonSerializable(typeof(RequestValidationResponse))]
+internal sealed partial class WorkerProxyJsonContext : JsonSerializerContext;

src/Functions.WorkerProxy/Properties/launchSettings.json

@@ -6,6 +6,7 @@
       "launchBrowser": false,
       "environmentVariables": {
         "ASPNETCORE_ENVIRONMENT": "Development",
+        "WORKERPROXY__PODNAME": "local-worker-pod",
         "WORKERPROXY__MANAGEMENTPORT": "8080",
         "WORKERPROXY__HTTPPORT": "28080",
         "WORKERPROXY__HTTPPROXYENDPOINT": "http://localhost:28080",