feat(worker): add claude-cli runner, queue service, and hub api
Runner stack (non-worktree path): IClaudeProcess + ClaudeProcess spawning the CLI with --output-format stream-json, prompt via stdin, parses the final type:"result" line into RunResult. LogWriter appends ndjson to ~/.todo-app/logs/<taskId>.ndjson. TaskRunner orchestrates DB transitions (MarkRunning -> MarkDone/Failed) and pushes TaskStarted/Message/Finished/ Updated via HubBroadcaster. Worktree-backed lists short-circuit with a "Slice E" failure message until git support lands. QueueService (BackgroundService) holds two in-memory slots (_queueSlot + _overrideSlot) guarded by a lock. Uses PeriodicTimer + SemaphoreSlim wake signal so WakeQueue() triggers an instant pickup. RunNow throws InvalidOperationException when override busy; CancelTask cancels the linked CTS which kills the child process tree. WorkerHub extended with GetActive, RunNow (translated to HubException variants), CancelTask, WakeQueue. HubBroadcaster exposes typed push methods. Tests: 26 pass (12 new). QueueServiceTests cover override-busy, schedule-filter, FIFO sequentiality, cancellation, plus a FakeClaudeProcess that blocks on a TCS for deterministic slot-state assertions. MessageParserTests cover result extraction + malformed/non-result lines. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Mika Kuns
278
tests/ClaudeDo.Worker.Tests/Services/QueueServiceTests.cs
Normal file
278
tests/ClaudeDo.Worker.Tests/Services/QueueServiceTests.cs
Normal file
@@ -0,0 +1,278 @@
|
||||
using ClaudeDo.Data.Models;
|
||||
using ClaudeDo.Data.Repositories;
|
||||
using ClaudeDo.Worker.Config;
|
||||
using ClaudeDo.Worker.Hub;
|
||||
using ClaudeDo.Worker.Runner;
|
||||
using ClaudeDo.Worker.Services;
|
||||
using ClaudeDo.Worker.Tests.Infrastructure;
|
||||
using Microsoft.AspNetCore.SignalR;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
|
||||
|
||||
namespace ClaudeDo.Worker.Tests.Services;
|
||||
|
||||
public sealed class QueueServiceTests : IDisposable
|
||||
{
|
||||
private readonly DbFixture _db = new();
|
||||
private readonly TaskRepository _taskRepo;
|
||||
private readonly ListRepository _listRepo;
|
||||
private readonly TagRepository _tagRepo;
|
||||
private readonly WorkerConfig _cfg;
|
||||
private readonly string _tempDir;
|
||||
|
||||
public QueueServiceTests()
|
||||
{
|
||||
_taskRepo = new TaskRepository(_db.Factory);
|
||||
_listRepo = new ListRepository(_db.Factory);
|
||||
_tagRepo = new TagRepository(_db.Factory);
|
||||
_tempDir = Path.Combine(Path.GetTempPath(), $"claudedo_test_{Guid.NewGuid():N}");
|
||||
Directory.CreateDirectory(_tempDir);
|
||||
_cfg = new WorkerConfig
|
||||
{
|
||||
SandboxRoot = Path.Combine(_tempDir, "sandbox"),
|
||||
LogRoot = Path.Combine(_tempDir, "logs"),
|
||||
QueueBackstopIntervalMs = 50, // fast for tests
|
||||
};
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
_db.Dispose();
|
||||
try { Directory.Delete(_tempDir, true); } catch { }
|
||||
}
|
||||
|
||||
private (QueueService service, FakeClaudeProcess fakeProcess) CreateService(
|
||||
Func<string, string, string, string, Func<string, Task>, CancellationToken, Task<RunResult>>? handler = null)
|
||||
{
|
||||
var fake = new FakeClaudeProcess(handler);
|
||||
var broadcaster = new HubBroadcaster(new FakeHubContext());
|
||||
var runner = new TaskRunner(fake, _taskRepo, _listRepo, broadcaster, _cfg,
|
||||
NullLogger<TaskRunner>.Instance);
|
||||
var service = new QueueService(_taskRepo, runner, _cfg, NullLogger<QueueService>.Instance);
|
||||
return (service, fake);
|
||||
}
|
||||
|
||||
private async Task<(string listId, long agentTagId)> SeedListWithAgentTag()
|
||||
{
|
||||
var listId = Guid.NewGuid().ToString();
|
||||
await _listRepo.AddAsync(new ListEntity { Id = listId, Name = "Test", CreatedAt = DateTime.UtcNow });
|
||||
|
||||
var tags = await _tagRepo.GetAllAsync();
|
||||
var agentTag = tags.First(t => t.Name == "agent");
|
||||
await _listRepo.AddTagAsync(listId, agentTag.Id);
|
||||
return (listId, agentTag.Id);
|
||||
}
|
||||
|
||||
private async Task<TaskEntity> SeedQueuedTask(string listId, DateTime? scheduledFor = null, DateTime? createdAt = null)
|
||||
{
|
||||
var task = new TaskEntity
|
||||
{
|
||||
Id = Guid.NewGuid().ToString(),
|
||||
ListId = listId,
|
||||
Title = "Test task",
|
||||
Description = "Do something",
|
||||
Status = TaskStatus.Queued,
|
||||
ScheduledFor = scheduledFor,
|
||||
CreatedAt = createdAt ?? DateTime.UtcNow,
|
||||
};
|
||||
await _taskRepo.AddAsync(task);
|
||||
return task;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RunNow_Throws_When_Override_Slot_Busy()
|
||||
{
|
||||
var (listId, _) = await SeedListWithAgentTag();
|
||||
var tcs = new TaskCompletionSource<RunResult>();
|
||||
|
||||
var (service, _) = CreateService((_, _, _, _, _, ct) => tcs.Task);
|
||||
|
||||
var task1 = await SeedQueuedTask(listId);
|
||||
var task2 = await SeedQueuedTask(listId);
|
||||
|
||||
await service.RunNow(task1.Id);
|
||||
|
||||
var ex = await Assert.ThrowsAsync<InvalidOperationException>(() => service.RunNow(task2.Id));
|
||||
Assert.Equal("override slot busy", ex.Message);
|
||||
|
||||
tcs.SetResult(new RunResult { ExitCode = 0, ResultMarkdown = "ok" });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RunNow_Throws_For_Unknown_Task()
|
||||
{
|
||||
var (service, _) = CreateService();
|
||||
await Assert.ThrowsAsync<KeyNotFoundException>(() => service.RunNow("nonexistent"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Schedule_Filter_Skips_Future_Tasks()
|
||||
{
|
||||
var (listId, _) = await SeedListWithAgentTag();
|
||||
await SeedQueuedTask(listId, scheduledFor: DateTime.UtcNow.AddHours(1));
|
||||
|
||||
var (service, fake) = CreateService((_, _, _, _, _, _) =>
|
||||
Task.FromResult(new RunResult { ExitCode = 0, ResultMarkdown = "ok" }));
|
||||
|
||||
using var cts = new CancellationTokenSource();
|
||||
|
||||
// Start the service loop, wake it, give it time.
|
||||
await service.StartAsync(cts.Token);
|
||||
service.WakeQueue();
|
||||
await Task.Delay(200);
|
||||
cts.Cancel();
|
||||
|
||||
// The fake should never have been called because the task is scheduled in the future.
|
||||
Assert.Equal(0, fake.CallCount);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Queue_FIFO_Sequentiality()
|
||||
{
|
||||
var (listId, _) = await SeedListWithAgentTag();
|
||||
|
||||
var order = new List<string>();
|
||||
var gate1 = new TaskCompletionSource();
|
||||
var gate2 = new TaskCompletionSource();
|
||||
var callCount = 0;
|
||||
|
||||
var (service, _) = CreateService(async (prompt, _, _, taskId, _, ct) =>
|
||||
{
|
||||
var n = Interlocked.Increment(ref callCount);
|
||||
lock (order) { order.Add(taskId); }
|
||||
if (n == 1) await gate1.Task;
|
||||
if (n == 2) gate2.SetResult();
|
||||
return new RunResult { ExitCode = 0, ResultMarkdown = "ok" };
|
||||
});
|
||||
|
||||
var task1 = await SeedQueuedTask(listId, createdAt: DateTime.UtcNow.AddSeconds(-2));
|
||||
var task2 = await SeedQueuedTask(listId, createdAt: DateTime.UtcNow.AddSeconds(-1));
|
||||
|
||||
using var cts = new CancellationTokenSource();
|
||||
await service.StartAsync(cts.Token);
|
||||
service.WakeQueue();
|
||||
|
||||
await Task.Delay(200);
|
||||
|
||||
// Only task1 should be running (task2 waiting).
|
||||
Assert.Single(order);
|
||||
Assert.Equal(task1.Id, order[0]);
|
||||
|
||||
// Release first task.
|
||||
gate1.SetResult();
|
||||
|
||||
// Wait for second task to complete.
|
||||
await gate2.Task.WaitAsync(TimeSpan.FromSeconds(5));
|
||||
|
||||
Assert.Equal(2, order.Count);
|
||||
Assert.Equal(task2.Id, order[1]);
|
||||
|
||||
cts.Cancel();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task CancelTask_Triggers_Cancellation()
|
||||
{
|
||||
var (listId, _) = await SeedListWithAgentTag();
|
||||
|
||||
var running = new TaskCompletionSource();
|
||||
var cancelled = false;
|
||||
|
||||
var (service, _) = CreateService(async (_, _, _, _, _, ct) =>
|
||||
{
|
||||
running.SetResult();
|
||||
try
|
||||
{
|
||||
await Task.Delay(Timeout.Infinite, ct);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
cancelled = true;
|
||||
throw;
|
||||
}
|
||||
return new RunResult { ExitCode = 0, ResultMarkdown = "ok" };
|
||||
});
|
||||
|
||||
var task = await SeedQueuedTask(listId);
|
||||
await service.RunNow(task.Id);
|
||||
|
||||
await running.Task.WaitAsync(TimeSpan.FromSeconds(5));
|
||||
|
||||
var result = service.CancelTask(task.Id);
|
||||
Assert.True(result);
|
||||
|
||||
await Task.Delay(200);
|
||||
Assert.True(cancelled);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task GetActive_Returns_Running_Slots()
|
||||
{
|
||||
var (listId, _) = await SeedListWithAgentTag();
|
||||
var tcs = new TaskCompletionSource<RunResult>();
|
||||
|
||||
var (service, _) = CreateService((_, _, _, _, _, _) => tcs.Task);
|
||||
|
||||
var task = await SeedQueuedTask(listId);
|
||||
await service.RunNow(task.Id);
|
||||
|
||||
var active = service.GetActive();
|
||||
Assert.Single(active);
|
||||
Assert.Equal("override", active[0].slot);
|
||||
Assert.Equal(task.Id, active[0].taskId);
|
||||
|
||||
tcs.SetResult(new RunResult { ExitCode = 0, ResultMarkdown = "ok" });
|
||||
}
|
||||
}
|
||||
|
||||
#region Test doubles
|
||||
|
||||
internal sealed class FakeClaudeProcess : IClaudeProcess
|
||||
{
|
||||
private readonly Func<string, string, string, string, Func<string, Task>, CancellationToken, Task<RunResult>> _handler;
|
||||
private int _callCount;
|
||||
|
||||
public int CallCount => _callCount;
|
||||
|
||||
public FakeClaudeProcess(
|
||||
Func<string, string, string, string, Func<string, Task>, CancellationToken, Task<RunResult>>? handler = null)
|
||||
{
|
||||
_handler = handler ?? ((_, _, _, _, _, _) =>
|
||||
Task.FromResult(new RunResult { ExitCode = 0, ResultMarkdown = "ok" }));
|
||||
}
|
||||
|
||||
public async Task<RunResult> RunAsync(string prompt, string workingDirectory, string logPath, string taskId,
|
||||
Func<string, Task> onStdoutLine, CancellationToken ct)
|
||||
{
|
||||
Interlocked.Increment(ref _callCount);
|
||||
return await _handler(prompt, workingDirectory, logPath, taskId, onStdoutLine, ct);
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class FakeHubContext : IHubContext<WorkerHub>
|
||||
{
|
||||
public IHubClients Clients { get; } = new FakeHubClients();
|
||||
public IGroupManager Groups => throw new NotImplementedException();
|
||||
}
|
||||
|
||||
internal sealed class FakeHubClients : IHubClients
|
||||
{
|
||||
private readonly FakeClientProxy _proxy = new();
|
||||
public IClientProxy All => _proxy;
|
||||
public IClientProxy AllExcept(IReadOnlyList<string> excludedConnectionIds) => _proxy;
|
||||
public IClientProxy Client(string connectionId) => _proxy;
|
||||
public IClientProxy Clients(IReadOnlyList<string> connectionIds) => _proxy;
|
||||
public IClientProxy Group(string groupName) => _proxy;
|
||||
public IClientProxy GroupExcept(string groupName, IReadOnlyList<string> excludedConnectionIds) => _proxy;
|
||||
public IClientProxy Groups(IReadOnlyList<string> groupNames) => _proxy;
|
||||
public IClientProxy User(string userId) => _proxy;
|
||||
public IClientProxy Users(IReadOnlyList<string> userIds) => _proxy;
|
||||
}
|
||||
|
||||
internal sealed class FakeClientProxy : IClientProxy
|
||||
{
|
||||
public Task SendCoreAsync(string method, object?[] args, CancellationToken cancellationToken = default) =>
|
||||
Task.CompletedTask;
|
||||
}
|
||||
|
||||
#endregion
|
||||
Reference in New Issue
Block a user