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

feat(worker): emit WorkerLog events from TaskRunner

Browse Source
This commit is contained in:
mika kuns
2026-04-23 14:46:10 +02:00
parent 0a7fcae137
commit ea4d2d7c0c
1 changed files with 21 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -49,7 +49,7 @@ public sealed class TaskRunner
list = await listRepo.GetByIdAsync(task.ListId, ct); list = await listRepo.GetByIdAsync(task.ListId, ct);
if (list is null) if (list is null)
{ {
await MarkFailed(task.Id, slot, "List not found."); await MarkFailed(task.Id, task.Title, slot, "List not found.");
return; return;
} }
listConfig = await listRepo.GetConfigAsync(task.ListId, ct); listConfig = await listRepo.GetConfigAsync(task.ListId, ct);
@@ -67,12 +67,13 @@ public sealed class TaskRunner
try try
{ {
wtCtx = await _wtManager.CreateAsync(task, list, ct); wtCtx = await _wtManager.CreateAsync(task, list, ct);
await _broadcaster.WorkerLog($"Created worktree for \"{task.Title}\"", WorkerLogLevel.Info, DateTime.UtcNow);
runDir = wtCtx.WorktreePath; runDir = wtCtx.WorktreePath;
} }
catch (Exception ex) catch (Exception ex)
{ {
_logger.LogError(ex, "Failed to create worktree for task {TaskId}", task.Id); _logger.LogError(ex, "Failed to create worktree for task {TaskId}", task.Id);
await MarkFailed(task.Id, slot, $"Worktree creation failed: {ex.Message}"); await MarkFailed(task.Id, task.Title, slot, $"Worktree creation failed: {ex.Message}");
return; return;
} }
} }
@@ -104,7 +105,7 @@ public sealed class TaskRunner
var prompt = sb.ToString(); var prompt = sb.ToString();
// Run 1. // Run 1.
var result = await RunOnceAsync(task.Id, slot, runDir, resolvedConfig, 1, false, prompt, ct); var result = await RunOnceAsync(task.Id, task.Title, slot, runDir, resolvedConfig, 1, false, prompt, ct);
if (result.IsSuccess) if (result.IsSuccess)
{ {
@@ -119,7 +120,7 @@ public sealed class TaskRunner
var retryConfig = resolvedConfig with { ResumeSessionId = result.SessionId }; var retryConfig = resolvedConfig with { ResumeSessionId = result.SessionId };
var retryPrompt = $"The previous attempt failed with:\n\n{result.ErrorMarkdown}\n\nTry again and fix the issues."; var retryPrompt = $"The previous attempt failed with:\n\n{result.ErrorMarkdown}\n\nTry again and fix the issues.";
var retryResult = await RunOnceAsync(task.Id, slot, runDir, retryConfig, 2, true, retryPrompt, ct); var retryResult = await RunOnceAsync(task.Id, task.Title, slot, runDir, retryConfig, 2, true, retryPrompt, ct);
if (retryResult.IsSuccess) if (retryResult.IsSuccess)
{ {
@@ -127,12 +128,12 @@ public sealed class TaskRunner
} }
else else
{ {
await HandleFailure(task.Id, slot, retryResult); await HandleFailure(task.Id, task.Title, slot, retryResult);
} }
} }
else else
{ {
await HandleFailure(task.Id, slot, result); await HandleFailure(task.Id, task.Title, slot, result);
} }
} }
@@ -141,12 +142,12 @@ public sealed class TaskRunner
catch (OperationCanceledException) catch (OperationCanceledException)
{ {
_logger.LogInformation("Task {TaskId} was cancelled", task.Id); _logger.LogInformation("Task {TaskId} was cancelled", task.Id);
await MarkFailed(task.Id, slot, "Task cancelled."); await MarkFailed(task.Id, task.Title, slot, "Task cancelled.");
} }
catch (Exception ex) catch (Exception ex)
{ {
_logger.LogError(ex, "Unhandled exception running task {TaskId}", task.Id); _logger.LogError(ex, "Unhandled exception running task {TaskId}", task.Id);
await MarkFailed(task.Id, slot, $"Unhandled error: {ex.Message}"); await MarkFailed(task.Id, task.Title, slot, $"Unhandled error: {ex.Message}");
} }
} }
@@ -204,7 +205,7 @@ public sealed class TaskRunner
await _broadcaster.TaskStarted(slot, taskId, now); await _broadcaster.TaskStarted(slot, taskId, now);
var nextRunNumber = lastRun.RunNumber + 1; var nextRunNumber = lastRun.RunNumber + 1;
var result = await RunOnceAsync(taskId, slot, runDir, resolvedConfig, nextRunNumber, false, followUpPrompt, ct); var result = await RunOnceAsync(taskId, task.Title, slot, runDir, resolvedConfig, nextRunNumber, false, followUpPrompt, ct);
if (result.IsSuccess) if (result.IsSuccess)
{ {
@@ -212,14 +213,14 @@ public sealed class TaskRunner
} }
else else
{ {
await HandleFailure(taskId, slot, result); await HandleFailure(taskId, task.Title, slot, result);
} }
await _broadcaster.TaskUpdated(taskId); await _broadcaster.TaskUpdated(taskId);
} }
private async Task<RunResult> RunOnceAsync( private async Task<RunResult> RunOnceAsync(
string taskId, string slot, string runDir, ClaudeRunConfig config, string taskId, string taskTitle, string slot, string runDir, ClaudeRunConfig config,
int runNumber, bool isRetry, string prompt, CancellationToken ct) int runNumber, bool isRetry, string prompt, CancellationToken ct)
{ {
var runId = Guid.NewGuid().ToString(); var runId = Guid.NewGuid().ToString();
@@ -250,6 +251,7 @@ public sealed class TaskRunner
try try
{ {
await _broadcaster.WorkerLog($"Started Claude for \"{taskTitle}\"", WorkerLogLevel.Info, DateTime.UtcNow);
var result = await _claude.RunAsync( var result = await _claude.RunAsync(
arguments, arguments,
prompt, prompt,
@@ -315,7 +317,10 @@ public sealed class TaskRunner
{ {
var committed = await _wtManager.CommitIfChangedAsync(wtCtx, task, list, ct); var committed = await _wtManager.CommitIfChangedAsync(wtCtx, task, list, ct);
if (committed) if (committed)
{
await _broadcaster.WorkerLog($"Committed changes in \"{task.Title}\"", WorkerLogLevel.Info, DateTime.UtcNow);
await _broadcaster.WorktreeUpdated(task.Id); await _broadcaster.WorktreeUpdated(task.Id);
}
} }
// Terminal DB write uses CancellationToken.None so the task status // Terminal DB write uses CancellationToken.None so the task status
@@ -327,12 +332,13 @@ public sealed class TaskRunner
var taskRepo = new TaskRepository(context); var taskRepo = new TaskRepository(context);
await taskRepo.MarkDoneAsync(task.Id, finishedAt, result.ResultMarkdown, CancellationToken.None); await taskRepo.MarkDoneAsync(task.Id, finishedAt, result.ResultMarkdown, CancellationToken.None);
} }
await _broadcaster.WorkerLog($"Finished \"{task.Title}\" (done)", WorkerLogLevel.Success, DateTime.UtcNow);
await _broadcaster.TaskFinished(slot, task.Id, "done", finishedAt); await _broadcaster.TaskFinished(slot, task.Id, "done", finishedAt);
_logger.LogInformation("Task {TaskId} completed (turns={Turns}, tokens_in={In}, tokens_out={Out})", _logger.LogInformation("Task {TaskId} completed (turns={Turns}, tokens_in={In}, tokens_out={Out})",
task.Id, result.TurnCount, result.TokensIn, result.TokensOut); task.Id, result.TurnCount, result.TokensIn, result.TokensOut);
} }
private async Task HandleFailure(string taskId, string slot, RunResult result) private async Task HandleFailure(string taskId, string taskTitle, string slot, RunResult result)
{ {
// Intentionally does not accept a CancellationToken: this is the // Intentionally does not accept a CancellationToken: this is the
// terminal write for a failed task and must always be persisted. // terminal write for a failed task and must always be persisted.
@@ -340,11 +346,12 @@ public sealed class TaskRunner
using var context = _dbFactory.CreateDbContext(); using var context = _dbFactory.CreateDbContext();
var taskRepo = new TaskRepository(context); var taskRepo = new TaskRepository(context);
await taskRepo.MarkFailedAsync(taskId, finishedAt, result.ErrorMarkdown, CancellationToken.None); await taskRepo.MarkFailedAsync(taskId, finishedAt, result.ErrorMarkdown, CancellationToken.None);
await _broadcaster.WorkerLog($"Finished \"{taskTitle}\" (failed)", WorkerLogLevel.Error, DateTime.UtcNow);
await _broadcaster.TaskFinished(slot, taskId, "failed", finishedAt); await _broadcaster.TaskFinished(slot, taskId, "failed", finishedAt);
_logger.LogWarning("Task {TaskId} failed (turns={Turns}): {Error}", taskId, result.TurnCount, result.ErrorMarkdown); _logger.LogWarning("Task {TaskId} failed (turns={Turns}): {Error}", taskId, result.TurnCount, result.ErrorMarkdown);
} }
private async Task MarkFailed(string taskId, string slot, string error) private async Task MarkFailed(string taskId, string taskTitle, string slot, string error)
{ {
try try
{ {
@@ -353,6 +360,7 @@ public sealed class TaskRunner
using var context = _dbFactory.CreateDbContext(); using var context = _dbFactory.CreateDbContext();
var taskRepo = new TaskRepository(context); var taskRepo = new TaskRepository(context);
await taskRepo.MarkFailedAsync(taskId, now, error, CancellationToken.None); await taskRepo.MarkFailedAsync(taskId, now, error, CancellationToken.None);
await _broadcaster.WorkerLog($"Finished \"{taskTitle}\" (failed)", WorkerLogLevel.Error, DateTime.UtcNow);
await _broadcaster.TaskFinished(slot, taskId, "failed", now); await _broadcaster.TaskFinished(slot, taskId, "failed", now);
await _broadcaster.TaskUpdated(taskId); await _broadcaster.TaskUpdated(taskId);
} }