feat(api): Admin work-item list, detail and live event stream (#279)

CLAUDE.md

@@ -199,7 +199,8 @@ Durable work for the Worker lives in `WorkItems` (#276, ADR 0008). Its kinds are
 - **Producing:** `IWorkQueue.EnqueueAsync` (`EfWorkQueue`, Core) is the only insert. A payload with a `DedupKey` coalesces: while an item with that key is `Queued`, enqueueing returns it (a partial unique index absorbs racing inserts).
 - **Scope:** a scoped item carries the `StorageConfigId` it belongs to (`IWorkPayload.StorageConfigId`). A per-user Worker (`WORKER__USER`) claims unscoped items and items of its user's configs; an all-user Worker claims everything. The served user is `WorkerState.ResolvedUserId`, read at each claim and tick, never cached at construction.
 - **Scheduling:** every Worker runs `WorkScheduler`. Each poll (`Worker:WorkPollIntervalSeconds`, 5 s) it recovers expired leases, then asks each `IWorkScheduleSource` for its schedules (a fixed `WorkSchedule<TPayload>` is its own source; `CrawlSchedules` returns one per active config the Worker serves) and whether each is due: nothing queued or running with its key and nothing finished within its interval. No timer state is kept, so overdue work runs right after downtime.
-- **Executing:** `WorkExecutor` claims the oldest queued item (`FOR UPDATE SKIP LOCKED`), one at a time, under a lease (`WorkLeaseSeconds`, 60 s) that a heartbeat (`WorkHeartbeatSeconds`, 2 s) extends while writing the progress snapshot. Every write about a running item is fenced by `(Id, Attempt)`. A fenced heartbeat cancels the handler and writes nothing more; a shutdown releases the item without a stall; a handler exception fails it with no retry. An expired lease requeues the item and counts a stall; the `WorkMaxStalls`th (3) fails it as stalled. An item handed back (release or recovery) while another with its key is already queued ends `Cancelled` as superseded, since the queued one does the same work. A Worker claims only the kinds it knows, so a newer kind waits for a newer Worker. `WorkItemRetention` deletes items finished more than `WorkItemRetentionDays` (30) ago.
+- **Executing:** `WorkExecutor` claims the oldest queued item (`FOR UPDATE SKIP LOCKED`), one at a time, under a lease (`WorkLeaseSeconds`, 60 s) that a heartbeat (`WorkHeartbeatSeconds`, 2 s) extends while writing the progress snapshot. `UpdatedAt` is the change cursor, so a heartbeat moves it only when the snapshot changed; `LeaseUntil` is what says the Worker is alive. Every write about a running item is fenced by `(Id, Attempt)`. A fenced heartbeat cancels the handler and writes nothing more; a shutdown releases the item without a stall; a handler exception fails it with no retry. An expired lease requeues the item and counts a stall; the `WorkMaxStalls`th (3) fails it as stalled. An item handed back (release or recovery) while another with its key is already queued ends `Cancelled` as superseded, since the queued one does the same work. A Worker claims only the kinds it knows, so a newer kind waits for a newer Worker. `WorkItemRetention` deletes items finished more than `WorkItemRetentionDays` (30) ago.
+- **Admin view (#279):** `GET /api/v1/admin/work-items` (filtered by status and kind, newest request first, paged), `.../{id}`, and `.../events`, a Server-Sent Events stream that re-queries `UpdatedAt` every `WorkItemEvents:IntervalSeconds` (2 s). An event's id is the cursor, and the list carries the cursor read before its page, so a client that lists and then subscribes with it as `Last-Event-ID` misses nothing; an item may arrive twice and is merged by id. The API reads items only through `IWorkItemReader` (Core), as `WorkItemView`. A result is shown once an item succeeded, an error once it failed or was cancelled.
 - **Adding a kind:** a `WorkKind` value, a payload record in Core implementing `IWorkPayload` (static `Kind`, optional `DedupKey`), an `IWorkHandler<TPayload>` in the Worker registered with `AddWorkHandler`, and for a periodic kind a `WorkSchedule<TPayload>` subclass registered as an `IWorkScheduleSource` (or a source that builds several). `WorkHandlerRegistry` fails startup unless every kind has exactly one payload and one handler; nothing switches on the kind.
 - **Handler contract:** delivery is at least once. Each run starts from scratch and converges, side effects are safe to repeat, and the returned summary becomes `Result`.
 

docs/schema.sql

@@ -95,7 +95,7 @@ COMMENT ON COLUMN "WorkItems"."Attempt" IS 'Incremented by every claim. Every wr
 COMMENT ON COLUMN "WorkItems"."Stalls" IS 'Lease expiries, counted by lease recovery. A graceful shutdown hands the item back without counting one; the item fails as stalled once the limit is reached.';
 COMMENT ON COLUMN "WorkItems"."ClaimedBy" IS 'The claiming Worker for display: its WORKER__USER or ''all'', then its host name.';
 COMMENT ON COLUMN "WorkItems"."LeaseUntil" IS 'Extended by every heartbeat. A Running item past it belongs to a dead or paused Worker and goes back to the queue.';
-COMMENT ON COLUMN "WorkItems"."UpdatedAt" IS 'Moves on every insert, claim, heartbeat and transition, so a reader can ask what changed since a point in time.';
+COMMENT ON COLUMN "WorkItems"."UpdatedAt" IS 'Moves on every insert, claim and transition, and on a heartbeat only when the progress snapshot changed, so a reader can ask what changed since a point in time.';
 COMMENT ON COLUMN "WorkItems"."ProgressNote" IS 'The latest progress snapshot (done, optional total, note), overwritten by each heartbeat; no history is kept.';
 COMMENT ON COLUMN "WorkItems"."Result" IS 'The handler''s results summary, written when the item succeeds.';
 

src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs

@@ -0,0 +1,122 @@
+using Anichron.API.Endpoints;
+using Anichron.API.Settings;
+using Anichron.API.Tests.Unit.Infrastructure;
+using Anichron.Core.Domain;
+using Anichron.Core.Work;
+using Anichron.Testing.Time;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Options;
+using Microsoft.Extensions.Time.Testing;
+using System.Net;
+using System.Text.Json;
+
+namespace Anichron.API.Tests.Unit.Endpoints;
+
+public sealed class WorkItemEndpointsHttpTests
+{
+    private static readonly Instant Start = Instant.FromUtc(2026, 10, 18, 9, 0);
+    private static readonly Guid AdminId = Guid.Parse("dddddddd-dddd-dddd-dddd-dddddddddddd");
+    private static readonly Guid ItemId = Guid.Parse("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa");
+
+    private readonly IWorkItemReader reader = Substitute.For<IWorkItemReader>();
+
+    private static CancellationToken Ct => TestContext.Current.CancellationToken;
+
+    private static WorkItemView Succeeded
+        => new(ItemId, WorkKind.Crawl, null, null, Start, WorkItemStatus.Succeeded, 1, 0, "all@host",
+            Start, Start + Duration.FromSeconds(9), Start + Duration.FromSeconds(9), new WorkProgress(8, 8, "8 found"),
+            /*lang=json,strict*/ """{"found": 8}""", null);
+
+    private Task<AuthenticatedTestApp> StartAsync()
+        => AuthenticatedTestApp.StartAsync(api => api.MapWorkItemEndpoints(), services =>
+        {
+            services.AddSingleton(reader);
+            services.AddSingleton<TimeProvider>(new FakeTimeProvider());
+            services.AddSingleton(Options.Create(new WorkItemEventsSettings()));
+            services.AddScoped<WorkItemEvents>();
+        });
+
+    [Fact]
+    public async Task List_BindsTheFiltersAndPagingFromTheQueryAsync()
+    {
+        reader.ListAsync(Arg.Any<WorkItemQuery>(), Arg.Any<CancellationToken>()).Returns(new WorkItemPage([Succeeded], 11));
+        await using var app = await StartAsync();
+        using var client = app.Client;
+
+        using var response = await client.SendAsync(
+            AuthenticatedTestApp.GetAsAdmin("/api/v1/admin/work-items?status=Succeeded&kind=Crawl&page=2&pageSize=10", AdminId), Ct);
+
+        response.StatusCode.Should().Be(HttpStatusCode.OK);
+        await reader.Received(1).ListAsync(
+            new WorkItemQuery(WorkItemStatus.Succeeded, WorkKind.Crawl, Skip: 10, Take: 10), Arg.Any<CancellationToken>());
+        using var body = JsonDocument.Parse(await response.Content.ReadAsStringAsync(Ct));
+        body.RootElement.GetProperty("total").GetInt32().Should().Be(11);
+        var item = body.RootElement.GetProperty("items")[0];
+        item.GetProperty("status").GetString().Should().Be("Succeeded");
+        item.GetProperty("result").GetProperty("found").GetInt32().Should().Be(8);
+    }
+
+    [Fact]
+    public async Task List_UnknownStatus_IsABadRequestAsync()
+    {
+        await using var app = await StartAsync();
+        using var client = app.Client;
+
+        using var response = await client.SendAsync(
+            AuthenticatedTestApp.GetAsAdmin("/api/v1/admin/work-items?status=Sleeping", AdminId), Ct);
+
+        response.StatusCode.Should().Be(HttpStatusCode.BadRequest);
+    }
+
+    [Theory]
+    [InlineData("/api/v1/admin/work-items")]
+    [InlineData("/api/v1/admin/work-items/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")]
+    [InlineData("/api/v1/admin/work-items/events")]
+    public async Task EveryRoute_ForANonAdmin_IsForbiddenAsync(string path)
+    {
+        await using var app = await StartAsync();
+        using var client = app.Client;
+
+        using var response = await client.SendAsync(AuthenticatedTestApp.Get(path, AdminId), Ct);
+
+        response.StatusCode.Should().Be(HttpStatusCode.Forbidden);
+    }
+
+    [Fact]
+    public async Task Events_ResumeFromLastEventIdAndFrameEachItemAsync()
+    {
+        var polls = 0;
+        reader.UpdatedSinceAsync(Arg.Any<Instant>(), Arg.Any<CancellationToken>())
+              .Returns(_ => ++polls == 1 ? [Succeeded] : Array.Empty<WorkItemView>());
+        await using var timeline = new Timeline();
+        await using var app = await StartAsync();
+        using var client = app.Client;
+        using var request = AuthenticatedTestApp.GetAsAdmin("/api/v1/admin/work-items/events", AdminId);
+        request.Headers.Add("Last-Event-ID", "2026-10-18T09:00:05Z");
+        using var response = await client.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, Ct);
+
+        timeline.Watch(Task.Run(async () => timeline.Record(Describe(await FirstFrameAsync(response))), CancellationToken.None));
+
+        await timeline.ExpectAsync(new Called(
+            "event: work-item | id: 2026-10-18T09:00:09Z | aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa 8 found"));
+        response.Content.Headers.ContentType!.MediaType.Should().Be("text/event-stream");
+        await reader.Received(1).UpdatedSinceAsync(Start - Duration.FromSeconds(5), Arg.Any<CancellationToken>());
+    }
+
+    private static string Describe(List<string> frame)
+    {
+        var data = frame.Single(line => line.StartsWith("data: ", StringComparison.Ordinal));
+        using var json = JsonDocument.Parse(data["data: ".Length..]);
+        var item = $"{json.RootElement.GetProperty("id").GetGuid()} {json.RootElement.GetProperty("progress").GetProperty("note").GetString()}";
+        return string.Join(" | ", frame.Where(line => line != data).Order(StringComparer.Ordinal).Append(item));
+    }
+
+    private static async Task<List<string>> FirstFrameAsync(HttpResponseMessage response)
+    {
+        using var stream = new StreamReader(await response.Content.ReadAsStreamAsync(Ct));
+        var lines = new List<string>();
+        while (await stream.ReadLineAsync(Ct) is { Length: > 0 } line)
+            lines.Add(line);
+        return lines;
+    }
+}