releases/ClaudeDo
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases 3 Wiki Activity

feat(worker,ui): wire EF Core into DI and update all consumers to IDbContextFactory

Browse Source 
Worker and App Program.cs: replace SqliteConnectionFactory+SchemaInitializer
with AddDbContextFactory<ClaudeDoDbContext> + Database.Migrate(). Repos
changed from AddSingleton to AddScoped.

All singleton services (QueueService, StaleTaskRecovery, WorktreeManager,
TaskRunner) and singleton ViewModels (MainWindowViewModel, TaskDetailViewModel,
TaskListViewModel, TaskEditorViewModel) now take IDbContextFactory<ClaudeDoDbContext>
and create short-lived contexts per operation.

Test infrastructure: DbFixture now uses EF migrations instead of SchemaInitializer;
all test classes create contexts via DbFixture.CreateContext().

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
mika kuns
2026-04-16 08:59:24 +02:00
parent b7be52a623
commit 36484ed45a
18 changed files with 479 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src/ClaudeDo.Worker/Program.cs
View File

@@ -5,6 +5,7 @@ using ClaudeDo.Worker.Config;
using ClaudeDo.Worker.Hub;
using ClaudeDo.Worker.Runner;
using ClaudeDo.Worker.Services;
using Microsoft.EntityFrameworkCore;
var cfg = WorkerConfig.Load();
@@ -14,18 +15,18 @@ var builder = WebApplication.CreateBuilder(args);
// doesn't think we crashed (~30s timeout). No-op when running interactively.
builder.Host.UseWindowsService(o => o.ServiceName = "ClaudeDoWorker");
// Initialize DB schema before the host starts accepting connections.
var dbFactory = new SqliteConnectionFactory(cfg.DbPath);
SchemaInitializer.Apply(dbFactory);
builder.Services.AddDbContextFactory<ClaudeDoDbContext>(opt =>
opt.UseSqlite($"Data Source={cfg.DbPath}"));
builder.Services.AddDbContext<ClaudeDoDbContext>(opt =>
opt.UseSqlite($"Data Source={cfg.DbPath}"));
builder.Services.AddSingleton(cfg);
builder.Services.AddSingleton(dbFactory);
builder.Services.AddSingleton<TagRepository>();
builder.Services.AddSingleton<ListRepository>();
builder.Services.AddSingleton<TaskRepository>();
builder.Services.AddSingleton<SubtaskRepository>();
builder.Services.AddSingleton<WorktreeRepository>();
builder.Services.AddSingleton<TaskRunRepository>();
builder.Services.AddScoped<TagRepository>();
builder.Services.AddScoped<ListRepository>();
builder.Services.AddScoped<TaskRepository>();
builder.Services.AddScoped<SubtaskRepository>();
builder.Services.AddScoped<WorktreeRepository>();
builder.Services.AddScoped<TaskRunRepository>();
builder.Services.AddHostedService<StaleTaskRecovery>();
builder.Services.AddSignalR();
@@ -51,6 +52,11 @@ builder.WebHost.UseUrls($"http://127.0.0.1:{cfg.SignalRPort}");
var app = builder.Build();
using (var scope = app.Services.CreateScope())
{
scope.ServiceProvider.GetRequiredService<ClaudeDoDbContext>().Database.Migrate();
}
app.MapHub<WorkerHub>("/hub");
app.Logger.LogInformation("ClaudeDo.Worker listening on http://127.0.0.1:{Port} (db: {Db})",

128
src/ClaudeDo.Worker/Runner/TaskRunner.cs
View File

@@ -1,18 +1,16 @@
using ClaudeDo.Data;
using ClaudeDo.Data.Models;
using ClaudeDo.Data.Repositories;
using ClaudeDo.Worker.Config;
using ClaudeDo.Worker.Hub;
using Microsoft.EntityFrameworkCore;
namespace ClaudeDo.Worker.Runner;
public sealed class TaskRunner
{
private readonly IClaudeProcess _claude;
private readonly TaskRepository _taskRepo;
private readonly TaskRunRepository _runRepo;
private readonly ListRepository _listRepo;
private readonly WorktreeRepository _wtRepo;
private readonly SubtaskRepository _subtaskRepo;
private readonly IDbContextFactory<ClaudeDoDbContext> _dbFactory;
private readonly HubBroadcaster _broadcaster;
private readonly WorktreeManager _wtManager;
private readonly ClaudeArgsBuilder _argsBuilder;
@@ -21,11 +19,7 @@ public sealed class TaskRunner
public TaskRunner(
IClaudeProcess claude,
TaskRepository taskRepo,
TaskRunRepository runRepo,
ListRepository listRepo,
WorktreeRepository wtRepo,
SubtaskRepository subtaskRepo,
IDbContextFactory<ClaudeDoDbContext> dbFactory,
HubBroadcaster broadcaster,
WorktreeManager wtManager,
ClaudeArgsBuilder argsBuilder,
@@ -33,11 +27,7 @@ public sealed class TaskRunner
ILogger<TaskRunner> logger)
{
_claude = claude;
_taskRepo = taskRepo;
_runRepo = runRepo;
_listRepo = listRepo;
_wtRepo = wtRepo;
_subtaskRepo = subtaskRepo;
_dbFactory = dbFactory;
_broadcaster = broadcaster;
_wtManager = wtManager;
_argsBuilder = argsBuilder;
@@ -49,11 +39,23 @@ public sealed class TaskRunner
{
try
{
var list = await _listRepo.GetByIdAsync(task.ListId, ct);
if (list is null)
ListEntity? list;
ListConfigEntity? listConfig;
List<SubtaskEntity> subtasks;
using (var context = _dbFactory.CreateDbContext())
{
await MarkFailed(task.Id, slot, "List not found.");
return;
var listRepo = new ListRepository(context);
list = await listRepo.GetByIdAsync(task.ListId, ct);
if (list is null)
{
await MarkFailed(task.Id, slot, "List not found.");
return;
}
listConfig = await listRepo.GetConfigAsync(task.ListId, ct);
var subtaskRepo = new SubtaskRepository(context);
subtasks = await subtaskRepo.GetByTaskIdAsync(task.Id, ct);
}
// Determine working directory: worktree or sandbox.
@@ -81,7 +83,6 @@ public sealed class TaskRunner
}
// Resolve config: task overrides > list config > null.
var listConfig = await _listRepo.GetConfigAsync(task.ListId, ct);
var resolvedConfig = new ClaudeRunConfig(
Model: task.Model ?? listConfig?.Model ?? "claude-sonnet-4-6",
SystemPrompt: task.SystemPrompt ?? listConfig?.SystemPrompt,
@@ -90,11 +91,14 @@ public sealed class TaskRunner
);
var now = DateTime.UtcNow;
await _taskRepo.MarkRunningAsync(task.Id, now, ct);
using (var context = _dbFactory.CreateDbContext())
{
var taskRepo = new TaskRepository(context);
await taskRepo.MarkRunningAsync(task.Id, now, ct);
}
await _broadcaster.TaskStarted(slot, task.Id, now);
// Build prompt.
var subtasks = await _subtaskRepo.GetByTaskIdAsync(task.Id, ct);
var sb = new System.Text.StringBuilder(task.Title);
if (!string.IsNullOrWhiteSpace(task.Description)) sb.Append("\n\n").Append(task.Description.Trim());
if (subtasks.Count > 0)
@@ -155,19 +159,34 @@ public sealed class TaskRunner
public async Task ContinueAsync(string taskId, string followUpPrompt, string slot, CancellationToken ct)
{
var task = await _taskRepo.GetByIdAsync(taskId, ct)
?? throw new KeyNotFoundException($"Task '{taskId}' not found.");
TaskEntity task;
TaskRunEntity lastRun;
ListEntity list;
ListConfigEntity? listConfig;
WorktreeEntity? worktree;
var lastRun = await _runRepo.GetLatestByTaskIdAsync(taskId, ct)
?? throw new InvalidOperationException("No previous run to continue.");
using (var context = _dbFactory.CreateDbContext())
{
var taskRepo = new TaskRepository(context);
task = await taskRepo.GetByIdAsync(taskId, ct)
?? throw new KeyNotFoundException($"Task '{taskId}' not found.");
if (lastRun.SessionId is null)
throw new InvalidOperationException("Previous run has no session ID — cannot resume.");
var runRepo = new TaskRunRepository(context);
lastRun = await runRepo.GetLatestByTaskIdAsync(taskId, ct)
?? throw new InvalidOperationException("No previous run to continue.");
var list = await _listRepo.GetByIdAsync(task.ListId, ct)
?? throw new InvalidOperationException("List not found.");
if (lastRun.SessionId is null)
throw new InvalidOperationException("Previous run has no session ID — cannot resume.");
var listRepo = new ListRepository(context);
list = await listRepo.GetByIdAsync(task.ListId, ct)
?? throw new InvalidOperationException("List not found.");
listConfig = await listRepo.GetConfigAsync(task.ListId, ct);
var wtRepo = new WorktreeRepository(context);
worktree = await wtRepo.GetByTaskIdAsync(taskId, ct);
}
var listConfig = await _listRepo.GetConfigAsync(task.ListId, ct);
var resolvedConfig = new ClaudeRunConfig(
Model: task.Model ?? listConfig?.Model,
SystemPrompt: task.SystemPrompt ?? listConfig?.SystemPrompt,
@@ -178,7 +197,6 @@ public sealed class TaskRunner
// Determine run directory from existing worktree or sandbox.
string runDir;
WorktreeContext? wtCtx = null;
var worktree = await _wtRepo.GetByTaskIdAsync(taskId, ct);
if (worktree is not null)
{
runDir = worktree.Path;
@@ -190,7 +208,11 @@ public sealed class TaskRunner
}
var now = DateTime.UtcNow;
await _taskRepo.MarkRunningAsync(taskId, now, ct);
using (var context = _dbFactory.CreateDbContext())
{
var taskRepo = new TaskRepository(context);
await taskRepo.MarkRunningAsync(taskId, now, ct);
}
await _broadcaster.TaskStarted(slot, taskId, now);
var nextRunNumber = lastRun.RunNumber + 1;
@@ -226,7 +248,12 @@ public sealed class TaskRunner
LogPath = logPath,
StartedAt = DateTime.UtcNow,
};
await _runRepo.AddAsync(run, ct);
using (var context = _dbFactory.CreateDbContext())
{
var runRepo = new TaskRunRepository(context);
await runRepo.AddAsync(run, ct);
}
var arguments = _argsBuilder.Build(config);
@@ -257,10 +284,15 @@ public sealed class TaskRunner
run.TokensIn = result.TokensIn;
run.TokensOut = result.TokensOut;
run.FinishedAt = DateTime.UtcNow;
await _runRepo.UpdateAsync(run, CancellationToken.None);
// Update denormalized fields on the task.
await _taskRepo.SetLogPathAsync(taskId, logPath, CancellationToken.None);
using (var context = _dbFactory.CreateDbContext())
{
var runRepo = new TaskRunRepository(context);
await runRepo.UpdateAsync(run, CancellationToken.None);
var taskRepo = new TaskRepository(context);
await taskRepo.SetLogPathAsync(taskId, logPath, CancellationToken.None);
}
return result;
}
@@ -273,8 +305,12 @@ public sealed class TaskRunner
run.FinishedAt = DateTime.UtcNow;
try
{
await _runRepo.UpdateAsync(run, CancellationToken.None);
await _taskRepo.SetLogPathAsync(taskId, logPath, CancellationToken.None);
using var context = _dbFactory.CreateDbContext();
var runRepo = new TaskRunRepository(context);
await runRepo.UpdateAsync(run, CancellationToken.None);
var taskRepo = new TaskRepository(context);
await taskRepo.SetLogPathAsync(taskId, logPath, CancellationToken.None);
}
catch (Exception updateEx)
{
@@ -297,7 +333,11 @@ public sealed class TaskRunner
// is never left as 'running' because of a cancel that arrived
// after the Claude run already succeeded.
var finishedAt = DateTime.UtcNow;
await _taskRepo.MarkDoneAsync(task.Id, finishedAt, result.ResultMarkdown, CancellationToken.None);
using (var context = _dbFactory.CreateDbContext())
{
var taskRepo = new TaskRepository(context);
await taskRepo.MarkDoneAsync(task.Id, finishedAt, result.ResultMarkdown, CancellationToken.None);
}
await _broadcaster.TaskFinished(slot, task.Id, "done", finishedAt);
_logger.LogInformation("Task {TaskId} completed (turns={Turns}, tokens_in={In}, tokens_out={Out})",
task.Id, result.TurnCount, result.TokensIn, result.TokensOut);
@@ -308,7 +348,9 @@ public sealed class TaskRunner
// Intentionally does not accept a CancellationToken: this is the
// terminal write for a failed task and must always be persisted.
var finishedAt = DateTime.UtcNow;
await _taskRepo.MarkFailedAsync(taskId, finishedAt, result.ErrorMarkdown, CancellationToken.None);
using var context = _dbFactory.CreateDbContext();
var taskRepo = new TaskRepository(context);
await taskRepo.MarkFailedAsync(taskId, finishedAt, result.ErrorMarkdown, CancellationToken.None);
await _broadcaster.TaskFinished(slot, taskId, "failed", finishedAt);
_logger.LogWarning("Task {TaskId} failed (turns={Turns}): {Error}", taskId, result.TurnCount, result.ErrorMarkdown);
}
@@ -319,7 +361,9 @@ public sealed class TaskRunner
{
var now = DateTime.UtcNow;
// Terminal write — never cancel.
await _taskRepo.MarkFailedAsync(taskId, now, error, CancellationToken.None);
using var context = _dbFactory.CreateDbContext();
var taskRepo = new TaskRepository(context);
await taskRepo.MarkFailedAsync(taskId, now, error, CancellationToken.None);
await _broadcaster.TaskFinished(slot, taskId, "failed", now);
await _broadcaster.TaskUpdated(taskId);
}

16
src/ClaudeDo.Worker/Runner/WorktreeManager.cs
View File

@@ -1,7 +1,9 @@
using ClaudeDo.Data;
using ClaudeDo.Data.Git;
using ClaudeDo.Data.Models;
using ClaudeDo.Data.Repositories;
using ClaudeDo.Worker.Config;
using Microsoft.EntityFrameworkCore;
namespace ClaudeDo.Worker.Runner;
@@ -10,14 +12,14 @@ public sealed record WorktreeContext(string WorktreePath, string BranchName, str
public sealed class WorktreeManager
{
private readonly GitService _git;
private readonly WorktreeRepository _wtRepo;
private readonly IDbContextFactory<ClaudeDoDbContext> _dbFactory;
private readonly WorkerConfig _cfg;
private readonly ILogger<WorktreeManager> _logger;
public WorktreeManager(GitService git, WorktreeRepository wtRepo, WorkerConfig cfg, ILogger<WorktreeManager> logger)
public WorktreeManager(GitService git, IDbContextFactory<ClaudeDoDbContext> dbFactory, WorkerConfig cfg, ILogger<WorktreeManager> logger)
{
_git = git;
_wtRepo = wtRepo;
_dbFactory = dbFactory;
_cfg = cfg;
_logger = logger;
}
@@ -50,7 +52,9 @@ public sealed class WorktreeManager
await _git.WorktreeAddAsync(workingDir, branchName, worktreePath, baseCommit, ct);
// Insert worktrees row AFTER git succeeds — if git throws, no row is created.
await _wtRepo.AddAsync(new WorktreeEntity
using var context = _dbFactory.CreateDbContext();
var wtRepo = new WorktreeRepository(context);
await wtRepo.AddAsync(new WorktreeEntity
{
TaskId = task.Id,
Path = worktreePath,
@@ -87,7 +91,9 @@ public sealed class WorktreeManager
var head = await _git.RevParseHeadAsync(ctx.WorktreePath, ct);
var diffStat = await _git.DiffStatAsync(ctx.WorktreePath, ctx.BaseCommit, head, ct);
await _wtRepo.UpdateHeadAsync(task.Id, head, diffStat, ct);
using var context = _dbFactory.CreateDbContext();
var wtRepo = new WorktreeRepository(context);
await wtRepo.UpdateHeadAsync(task.Id, head, diffStat, ct);
_logger.LogInformation("Committed changes for task {TaskId}: {Head}", task.Id, head);
return true;

23
src/ClaudeDo.Worker/Services/QueueService.cs
View File

@@ -1,7 +1,9 @@
using ClaudeDo.Data;
using ClaudeDo.Data.Models;
using ClaudeDo.Data.Repositories;
using ClaudeDo.Worker.Config;
using ClaudeDo.Worker.Runner;
using Microsoft.EntityFrameworkCore;
namespace ClaudeDo.Worker.Services;
@@ -14,7 +16,7 @@ public sealed class QueueSlotState
public sealed class QueueService : BackgroundService
{
private readonly TaskRepository _taskRepo;
private readonly IDbContextFactory<ClaudeDoDbContext> _dbFactory;
private readonly TaskRunner _runner;
private readonly WorkerConfig _cfg;
private readonly ILogger<QueueService> _logger;
@@ -26,12 +28,12 @@ public sealed class QueueService : BackgroundService
private readonly SemaphoreSlim _wakeSignal = new(0, 1);
public QueueService(
TaskRepository taskRepo,
IDbContextFactory<ClaudeDoDbContext> dbFactory,
TaskRunner runner,
WorkerConfig cfg,
ILogger<QueueService> logger)
{
_taskRepo = taskRepo;
_dbFactory = dbFactory;
_runner = runner;
_cfg = cfg;
_logger = logger;
@@ -56,7 +58,9 @@ public sealed class QueueService : BackgroundService
public async Task RunNow(string taskId)
{
var task = await _taskRepo.GetByIdAsync(taskId);
using var context = _dbFactory.CreateDbContext();
var taskRepo = new TaskRepository(context);
var task = await taskRepo.GetByIdAsync(taskId);
if (task is null)
throw new KeyNotFoundException($"Task '{taskId}' not found.");
@@ -78,7 +82,9 @@ public sealed class QueueService : BackgroundService
public async Task<string> ContinueTask(string taskId, string followUpPrompt)
{
var task = await _taskRepo.GetByIdAsync(taskId)
using var context = _dbFactory.CreateDbContext();
var taskRepo = new TaskRepository(context);
var task = await taskRepo.GetByIdAsync(taskId)
?? throw new KeyNotFoundException($"Task '{taskId}' not found.");
if (task.Status == Data.Models.TaskStatus.Running)
@@ -144,7 +150,12 @@ public sealed class QueueService : BackgroundService
if (_queueSlot is not null) continue;
var task = await _taskRepo.GetNextQueuedAgentTaskAsync(DateTime.UtcNow, stoppingToken);
TaskEntity? task;
using (var context = _dbFactory.CreateDbContext())
{
var taskRepo = new TaskRepository(context);
task = await taskRepo.GetNextQueuedAgentTaskAsync(DateTime.UtcNow, stoppingToken);
}
if (task is null) continue;
lock (_lock)

12
src/ClaudeDo.Worker/Services/StaleTaskRecovery.cs
View File

@@ -1,21 +1,25 @@
using ClaudeDo.Data;
using ClaudeDo.Data.Repositories;
using Microsoft.EntityFrameworkCore;
namespace ClaudeDo.Worker.Services;
public sealed class StaleTaskRecovery : IHostedService
{
private readonly TaskRepository _tasks;
private readonly IDbContextFactory<ClaudeDoDbContext> _dbFactory;
private readonly ILogger<StaleTaskRecovery> _logger;
public StaleTaskRecovery(TaskRepository tasks, ILogger<StaleTaskRecovery> logger)
public StaleTaskRecovery(IDbContextFactory<ClaudeDoDbContext> dbFactory, ILogger<StaleTaskRecovery> logger)
{
_tasks = tasks;
_dbFactory = dbFactory;
_logger = logger;
}
public async Task StartAsync(CancellationToken cancellationToken)
{
var flipped = await _tasks.FlipAllRunningToFailedAsync("worker restart", cancellationToken);
using var context = _dbFactory.CreateDbContext();
var tasks = new TaskRepository(context);
var flipped = await tasks.FlipAllRunningToFailedAsync("worker restart", cancellationToken);
if (flipped > 0)
_logger.LogWarning("Stale task recovery: flipped {Count} running task(s) to failed", flipped);
else