Add /work /heartbeat /gradient /weights endpoints + weight store

Phase D-1 step 2d. Lights up the rest of the worker-facing REST
surface the distributed training loop needs, plus the filesystem
weight-versioned blob store that backs /weights.

FileSystemWeightStore:
- Flat directory with v{version:D10}.bin + v{version:D10}.bin.sha256
  sidecar files. SaveVersion writes atomically through a .tmp
  staging file + rename so a crash mid-write cannot expose a
  half-written blob. TryGetManifest + TryOpenReadStream serve
  workers without any in-memory caching. ListVersions returns
  versions in ascending numeric order; GetLatestVersion returns
  the highest. Refuses to overwrite an existing version —
  coordinator weight versions are immutable by contract.
- 9 xunit tests locking empty-store behavior, save + round trip,
  overwrite rejection, manifest consistency, stream contents,
  ordering, and auto-directory-creation.

Worker API endpoints (all JWT-guarded by BitNetWorkerPolicy):
- GET  /work
    Reads client_id from JWT, calls SqliteWorkQueueStore
    .TryClaimNextPending with a lease = 2x the target task
    duration (safety margin over the 10-minute goal), and maps
    the WorkTaskRecord to a WorkTaskAssignment DTO with a
    fully-qualified {BaseUrl}/weights/{version} WeightUrl so
    the worker knows where to download the weight blob.
    Returns 204 No Content when the queue is empty so workers
    can long-poll without allocating an object per empty poll.
- POST /heartbeat
    Validates body + client_id, calls
    SqliteWorkerRegistryStore.TouchHeartbeat. Returns
    HeartbeatResponse with ShouldDrain=false for now; the
    coordinator dashboard will set it later when the operator
    requests a rolling drain. Returns 410 Gone for an unknown
    worker so the worker knows to re-/register.
- POST /gradient
    D-1 stub: validates that the submission's worker_id matches
    the JWT client_id, then calls SqliteWorkQueueStore
    .MarkCompleted to flip the task Done. Logs the gradient
    format + payload size + tokens seen + loss + staleness for
    visibility. Does NOT yet decode the gradient or apply it to
    the global weight copy — that lands in the D-4 commit that
    introduces the int8-ef gradient decoder.
- GET  /weights/{version:long}
    Uses FileSystemWeightStore.TryOpenReadStream to stream the
    blob with Results.File + enableRangeProcessing so workers
    over ngrok can resume partial downloads.

Program.cs wires FileSystemWeightStore as a DI singleton whose
root directory is derived from the database path (sibling
`weights/` directory).

Fast-lane regression 232/232 passing on net10 slice; no new
tests for the endpoints yet — they need a TestServer harness
that covers the full JWT issuance flow end-to-end, which lands
together with the Worker HttpClient wire-up.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: sharpninja

src/BitNetSharp.Distributed.Coordinator/Persistence/FileSystemWeightStore.cs

@@ -0,0 +1,197 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Security.Cryptography;
+using System.Text;
+
+namespace BitNetSharp.Distributed.Coordinator.Persistence;
+
+/// <summary>
+/// Filesystem-backed versioned blob store for the coordinator's
+/// global training weights. The work queue stores
+/// <c>weight_version</c> integers only; actual ternary-packed model
+/// bytes live on disk under a flat directory so workers can stream
+/// them via <c>GET /weights/{version}</c> with range support.
+///
+/// <para>
+/// Layout under <c>{rootDirectory}</c>:
+/// <code>
+///   v0000000001.bin
+///   v0000000001.sha256
+///   v0000000002.bin
+///   v0000000002.sha256
+///   ...
+/// </code>
+/// Each <c>.bin</c> file is the immutable blob payload; the matching
+/// <c>.sha256</c> file holds the hex digest the coordinator computed
+/// at save time so later readers can verify integrity without
+/// rehashing.
+/// </para>
+/// </summary>
+public sealed class FileSystemWeightStore
+{
+    private readonly string _rootDirectory;
+    private readonly object _writeGate = new();
+
+    public FileSystemWeightStore(string rootDirectory)
+    {
+        ArgumentException.ThrowIfNullOrWhiteSpace(rootDirectory);
+        _rootDirectory = rootDirectory;
+        Directory.CreateDirectory(_rootDirectory);
+    }
+
+    /// <summary>
+    /// Persists the given blob as the contents of the specified
+    /// weight version. Overwriting an existing version is disallowed
+    /// — the whole point of versioning is that each
+    /// <c>weight_version</c> is immutable.
+    /// </summary>
+    public WeightVersionManifest SaveVersion(long version, ReadOnlySpan<byte> payload)
+    {
+        if (version <= 0)
+        {
+            throw new ArgumentOutOfRangeException(nameof(version), "Weight version must be positive.");
+        }
+
+        var binPath = PathForVersion(version);
+        var shaPath = binPath + ".sha256";
+
+        lock (_writeGate)
+        {
+            if (File.Exists(binPath))
+            {
+                throw new InvalidOperationException(
+                    $"Weight version {version} already exists at {binPath}. Coordinator weight versions are immutable.");
+            }
+
+            // Write payload atomically via a staging file then rename
+            // so a crash mid-write cannot leave a half-written version
+            // visible to /weights/{version} readers.
+            var stagingPath = binPath + ".tmp";
+            try
+            {
+                using (var stream = File.Create(stagingPath))
+                {
+                    stream.Write(payload);
+                }
+
+                File.Move(stagingPath, binPath);
+
+                var hash = Sha256Hex(payload);
+                File.WriteAllText(shaPath, hash, Encoding.ASCII);
+                return new WeightVersionManifest(version, binPath, payload.Length, hash);
+            }
+            catch
+            {
+                if (File.Exists(stagingPath))
+                {
+                    File.Delete(stagingPath);
+                }
+
+                throw;
+            }
+        }
+    }
+
+    /// <summary>
+    /// Returns the manifest (path, size, hash) for the requested
+    /// version if it exists on disk, or <c>null</c> otherwise.
+    /// </summary>
+    public WeightVersionManifest? TryGetManifest(long version)
+    {
+        if (version <= 0)
+        {
+            return null;
+        }
+
+        var binPath = PathForVersion(version);
+        var shaPath = binPath + ".sha256";
+        if (!File.Exists(binPath) || !File.Exists(shaPath))
+        {
+            return null;
+        }
+
+        var length = new FileInfo(binPath).Length;
+        var hash = File.ReadAllText(shaPath, Encoding.ASCII).Trim();
+        return new WeightVersionManifest(version, binPath, length, hash);
+    }
+
+    /// <summary>
+    /// Opens a read-only stream over the requested version's blob,
+    /// or <c>null</c> if the version does not exist on disk. Caller
+    /// owns the returned stream and must dispose it.
+    /// </summary>
+    public Stream? TryOpenReadStream(long version)
+    {
+        var manifest = TryGetManifest(version);
+        if (manifest is null)
+        {
+            return null;
+        }
+
+        return new FileStream(
+            manifest.PhysicalPath,
+            FileMode.Open,
+            FileAccess.Read,
+            FileShare.Read,
+            bufferSize: 64 * 1024,
+            useAsync: true);
+    }
+
+    /// <summary>
+    /// Enumerates every weight version currently on disk in ascending
+    /// numeric order. Handy for the /status dashboard.
+    /// </summary>
+    public IReadOnlyList<long> ListVersions()
+    {
+        return Directory
+            .EnumerateFiles(_rootDirectory, "v*.bin")
+            .Select(path => Path.GetFileNameWithoutExtension(path))
+            .Select(name => long.TryParse(name.TrimStart('v'), out var parsed) ? parsed : 0L)
+            .Where(v => v > 0)
+            .OrderBy(v => v)
+            .ToList();
+    }
+
+    /// <summary>
+    /// Returns the highest-numbered version currently on disk, or
+    /// <c>null</c> if the store is empty.
+    /// </summary>
+    public long? GetLatestVersion()
+    {
+        var versions = ListVersions();
+        return versions.Count == 0 ? null : versions[^1];
+    }
+
+    /// <summary>
+    /// Computes the filesystem path for the given version using a
+    /// 10-digit zero-padded name so alphabetical enumeration matches
+    /// numerical order for the first ten billion versions.
+    /// </summary>
+    private string PathForVersion(long version) =>
+        Path.Combine(_rootDirectory, $"v{version:D10}.bin");
+
+    private static string Sha256Hex(ReadOnlySpan<byte> payload)
+    {
+        Span<byte> hash = stackalloc byte[32];
+        SHA256.HashData(payload, hash);
+        var sb = new StringBuilder(hash.Length * 2);
+        foreach (var b in hash)
+        {
+            sb.Append(b.ToString("x2", System.Globalization.CultureInfo.InvariantCulture));
+        }
+        return sb.ToString();
+    }
+}
+
+/// <summary>
+/// Summary of a weight version on disk. Returned by
+/// <see cref="FileSystemWeightStore.SaveVersion"/> and
+/// <see cref="FileSystemWeightStore.TryGetManifest"/>.
+/// </summary>
+public sealed record WeightVersionManifest(
+    long Version,
+    string PhysicalPath,
+    long SizeBytes,
+    string Sha256Hex);

src/BitNetSharp.Distributed.Coordinator/Program.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.IO;
 using System.Linq;
 using System.Security.Claims;
 using BitNetSharp.Distributed.Contracts;
@@ -79,6 +80,14 @@
     var time  = sp.GetRequiredService<TimeProvider>();
     return new SqliteClientRevocationStore(BuildConnectionString(coord), time);
 });
+builder.Services.AddSingleton(sp =>
+{
+    var coord = sp.GetRequiredService<CoordinatorOptions>();
+    var weightsDir = Path.Combine(
+        Path.GetDirectoryName(Path.GetFullPath(coord.DatabasePath)) ?? ".",
+        "weights");
+    return new FileSystemWeightStore(weightsDir);
+});
 
 // ── Worker client registry + Duende IdentityServer ────────────────
 var workerRegistry = new WorkerClientRegistry();
@@ -200,10 +209,11 @@
 
 var app = builder.Build();
 
-// Ensure all three stores create their schema on startup.
+// Ensure all stores create their schema / directories on startup.
 _ = app.Services.GetRequiredService<SqliteWorkQueueStore>();
 _ = app.Services.GetRequiredService<SqliteWorkerRegistryStore>();
 _ = app.Services.GetRequiredService<SqliteClientRevocationStore>();
+_ = app.Services.GetRequiredService<FileSystemWeightStore>();
 
 app.UseAuthentication();
 app.UseAuthorization();
@@ -295,6 +305,159 @@
     return Results.Ok(response);
 }).RequireAuthorization(IdentityServerResources.WorkerPolicyName);
 
+// ── /work — claim the next pending task for this worker ──────────
+app.MapGet("/work", (
+    HttpContext http,
+    SqliteWorkQueueStore workQueue,
+    CoordinatorOptions options) =>
+{
+    var clientId = http.User.FindFirst("client_id")?.Value;
+    if (string.IsNullOrWhiteSpace(clientId))
+    {
+        return Results.Json(
+            new ErrorResponse("unknown_client", "JWT did not carry a client_id claim."),
+            statusCode: StatusCodes.Status401Unauthorized);
+    }
+
+    var leaseDuration = TimeSpan.FromSeconds(options.TargetTaskDurationSeconds * 2);
+    var claimed = workQueue.TryClaimNextPending(clientId, leaseDuration);
+    if (claimed is null)
+    {
+        return Results.StatusCode(StatusCodes.Status204NoContent);
+    }
+
+    var baseUrl = options.BaseUrl.TrimEnd('/');
+    var assignment = new WorkTaskAssignment(
+        TaskId: claimed.TaskId,
+        WeightVersion: claimed.WeightVersion,
+        WeightUrl: $"{baseUrl}/weights/{claimed.WeightVersion}",
+        ShardId: claimed.ShardId,
+        ShardOffset: claimed.ShardOffset,
+        ShardLength: claimed.ShardLength,
+        TokensPerTask: claimed.TokensPerTask,
+        KLocalSteps: claimed.KLocalSteps,
+        HyperparametersJson: claimed.HyperparametersJson,
+        DeadlineUtc: claimed.DeadlineUtc ?? DateTimeOffset.UtcNow.Add(leaseDuration));
+
+    return Results.Ok(assignment);
+}).RequireAuthorization(IdentityServerResources.WorkerPolicyName);
+
+// ── /heartbeat — worker pings the coordinator periodically ───────
+app.MapPost("/heartbeat", (
+    [FromBody] HeartbeatRequest request,
+    HttpContext http,
+    SqliteWorkerRegistryStore workerStore,
+    CoordinatorOptions options,
+    TimeProvider time) =>
+{
+    if (request is null)
+    {
+        return Results.Json(
+            new ErrorResponse("invalid_request", "Heartbeat body is missing."),
+            statusCode: StatusCodes.Status400BadRequest);
+    }
+
+    var clientId = http.User.FindFirst("client_id")?.Value;
+    if (string.IsNullOrWhiteSpace(clientId))
+    {
+        return Results.Json(
+            new ErrorResponse("unknown_client", "JWT did not carry a client_id claim."),
+            statusCode: StatusCodes.Status401Unauthorized);
+    }
+
+    var touched = workerStore.TouchHeartbeat(clientId);
+    if (!touched)
+    {
+        return Results.Json(
+            new ErrorResponse("unregistered", "Worker must POST /register before heartbeating."),
+            statusCode: StatusCodes.Status410Gone);
+    }
+
+    return Results.Ok(new HeartbeatResponse(
+        ShouldDrain: false,
+        RecommendedTokensPerTaskOverride: null,
+        ServerTime: time.GetUtcNow()));
+}).RequireAuthorization(IdentityServerResources.WorkerPolicyName);
+
+// ── /gradient — worker reports task completion ──────────────────
+// D-1 stub: validates ownership, marks the task Done, does NOT yet
+// apply the gradient to the global weights. Phase D-4 introduces the
+// gradient decoder + weight updater.
+app.MapPost("/gradient", (
+    [FromBody] GradientSubmission submission,
+    HttpContext http,
+    SqliteWorkQueueStore workQueue,
+    ILogger<Program> logger) =>
+{
+    if (submission is null)
+    {
+        return Results.Json(
+            new ErrorResponse("invalid_request", "Gradient body is missing."),
+            statusCode: StatusCodes.Status400BadRequest);
+    }
+
+    var clientId = http.User.FindFirst("client_id")?.Value;
+    if (string.IsNullOrWhiteSpace(clientId) || clientId != submission.WorkerId)
+    {
+        return Results.Json(
+            new ErrorResponse("worker_mismatch", "Gradient workerId must match the JWT client_id."),
+            statusCode: StatusCodes.Status403Forbidden);
+    }
+
+    var completed = workQueue.MarkCompleted(submission.TaskId, clientId);
+    if (!completed)
+    {
+        return Results.Json(
+            new ErrorResponse("task_not_assigned", "Task is not currently assigned to this worker."),
+            statusCode: StatusCodes.Status409Conflict);
+    }
+
+    logger.LogInformation(
+        "Accepted gradient for task {TaskId} from worker {ClientId}: format={Format}, bytes={Size}, tokens={Tokens}, loss={Loss}, staleness={Staleness}",
+        submission.TaskId,
+        clientId,
+        submission.GradientFormat,
+        submission.GradientPayload?.Length ?? 0,
+        submission.TokensSeen,
+        submission.LossAfter,
+        0);
+
+    return Results.Ok(new
+    {
+        accepted = true,
+        task_id = submission.TaskId,
+        worker_id = clientId
+    });
+}).RequireAuthorization(IdentityServerResources.WorkerPolicyName);
+
+// ── /weights/{version} — streams a weight blob to the worker ─────
+app.MapGet("/weights/{version:long}", (
+    long version,
+    FileSystemWeightStore weights) =>
+{
+    var manifest = weights.TryGetManifest(version);
+    if (manifest is null)
+    {
+        return Results.Json(
+            new ErrorResponse("unknown_version", $"Weight version {version} is not available."),
+            statusCode: StatusCodes.Status404NotFound);
+    }
+
+    var stream = weights.TryOpenReadStream(version);
+    if (stream is null)
+    {
+        return Results.Json(
+            new ErrorResponse("unknown_version", $"Weight version {version} is not available."),
+            statusCode: StatusCodes.Status404NotFound);
+    }
+
+    return Results.File(
+        fileStream: stream,
+        contentType: "application/octet-stream",
+        fileDownloadName: $"bitnet-weights-v{version}.bin",
+        enableRangeProcessing: true);
+}).RequireAuthorization(IdentityServerResources.WorkerPolicyName);
+
 // ── Admin Blazor UI (cookie + OIDC) ───────────────────────────────
 app.MapRazorComponents<App>();