Teamup/tests/TeamUp.IntegrationTests/AgentRunQueueTests.cs

using System.Net;
using System.Net.Http.Headers;
using System.Net.Http.Json;
using Microsoft.Extensions.DependencyInjection;
using TeamUp.Modules.Assembler.Queue;
using TeamUp.Modules.Assembler.Runtime;
using Xunit;

namespace TeamUp.IntegrationTests;

/// <summary>
/// M4 Increment 1: a run is enqueued (web), claimed via FOR UPDATE SKIP LOCKED + processed
/// (worker services), and the AgentRun lifecycle reaches Completed.
/// </summary>
public sealed class AgentRunQueueTests(PostgresFixture postgres) : IClassFixture<PostgresFixture>
{
    private sealed record BootstrapResponse(string Token, Guid MemberId, Guid OrganizationId);

    private sealed record RunResponse(
        Guid Id, Guid SeatId, Guid WorkItemId, Guid? AgentId, string Status,
        string? ActionType, string? ActionRisk, string? Prompt, string? Output, string? Error);

    [Fact]
    public async Task Run_is_enqueued_claimed_and_processed()
    {
        await using var factory = new TeamUpWebFactory(postgres.ConnectionString);
        using var anon = factory.CreateClient();

        var bootstrap = await anon.PostAsJsonAsync("/api/identity/bootstrap", new
        {
            organizationName = "AliaSaaS",
            ownerEmail = "owner@alia.test",
            ownerDisplayName = "Owner",
            ownerPassword = "Passw0rd!",
        });
        var owner = await bootstrap.Content.ReadFromJsonAsync<BootstrapResponse>();
        Assert.NotNull(owner);

        using var client = factory.CreateClient();
        client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", owner!.Token);

        // Enqueue a run (web).
        var created = await client.PostAsJsonAsync("/api/assembler/runs", new
        {
            seatId = Guid.NewGuid(),
            workItemId = Guid.NewGuid(),
        });
        Assert.Equal(HttpStatusCode.OK, created.StatusCode);
        var run = await created.Content.ReadFromJsonAsync<RunResponse>();
        Assert.Equal("Queued", run!.Status);

        // Drain one job exactly as the worker does: claim (SKIP LOCKED) then process.
        await using (var scope = factory.Services.CreateAsyncScope())
        {
            var queue = scope.ServiceProvider.GetRequiredService<JobQueue>();
            var job = await queue.ClaimNextAsync("test-worker");
            Assert.NotNull(job);

            var executor = scope.ServiceProvider.GetRequiredService<AgentRunExecutor>();
            await executor.ProcessAsync(job!);
        }

        // The next claim finds nothing (the only job is done).
        await using (var scope = factory.Services.CreateAsyncScope())
        {
            var queue = scope.ServiceProvider.GetRequiredService<JobQueue>();
            Assert.Null(await queue.ClaimNextAsync("test-worker"));
        }

        // With no agent configured for the random seat, the run reaches a terminal Failed state
        // (the full assemble→model→parse path is covered by AssemblerRunTests).
        var fetched = await client.GetFromJsonAsync<RunResponse>($"/api/assembler/runs/{run.Id}");
        Assert.Equal("Failed", fetched!.Status);
        Assert.Contains("not found", fetched.Error ?? string.Empty, StringComparison.OrdinalIgnoreCase);
    }
}