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

feat(mission-control): show running Worker operations in a Running now section

Browse Source 
Registry of active verify/merge/rebase/worktree-creation operations in the Worker,
GetActiveOperations snapshot + OperationStarted/OperationEnded hub events, and a
Running now section in Mission Control with ticking elapsed time.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
mika kuns
2026-10-01 12:08:43 +02:00
co-authored by Claude Sonnet 5.5
parent 488c28a9d5
commit 2dbec543df
18 changed files with 539 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ClaudeDo.Worker/Hub/HubBroadcaster.cs
+8
View File
@@ -78,6 +78,14 @@ public sealed class HubBroadcaster : IPrimeBroadcaster, IRefineBroadcaster
public Task OperationProgress(string opKey, string phase, int current, int total) =>
_hub.Clients.All.SendAsync("OperationProgress", opKey, phase, current, total);
/// <summary>A tracked Worker operation (verify, merge, rebase, worktree creation) began.</summary>
public Task OperationStarted(RunningOperationDto operation) =>
_hub.Clients.All.SendAsync("OperationStarted", operation);
/// <summary>The tracked operation with this id finished, failed or threw.</summary>
public Task OperationEnded(string operationId) =>
_hub.Clients.All.SendAsync("OperationEnded", operationId);
public Task PlanningMergeStarted(string planningTaskId, string targetBranch) =>
_hub.Clients.All.SendAsync("PlanningMergeStarted", planningTaskId, targetBranch);
src/ClaudeDo.Worker/Hub/WorkerHub.cs
+8 -1
View File
@@ -75,6 +75,7 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
private readonly HandlerRestartCoordinator? _handlerRestart;
private readonly HandlerRoundState? _roundState;
private readonly ConnectedClientTracker? _clientTracker;
private readonly RunningOperationRegistry? _operations;
public WorkerHub(
QueueService queue,
@@ -116,7 +117,8 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
TicketImportService ticketImport = null!,
HandlerRestartCoordinator? handlerRestart = null,
HandlerRoundState? roundState = null,
ConnectedClientTracker? clientTracker = null)
ConnectedClientTracker? clientTracker = null,
RunningOperationRegistry? operations = null)
{
_queue = queue;
_waker = waker;
@@ -158,6 +160,7 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
_handlerRestart = handlerRestart;
_roundState = roundState;
_clientTracker = clientTracker;
_operations = operations;
}
// Tracks connected UI clients so an out-of-band caller (external MCP's start_list_handler)
@@ -203,6 +206,10 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
public IReadOnlyList<WorkerLogRecord> GetRecentLogs() =>
_logBuffer?.Snapshot() ?? Array.Empty<WorkerLogRecord>();
/// <summary>Operations running right now — lets a connecting or reconnecting UI recover state before the started/ended events.</summary>
public IReadOnlyList<RunningOperationDto> GetActiveOperations() =>
_operations?.Snapshot() ?? Array.Empty<RunningOperationDto>();
// Maps the two exceptions service methods throw into client-facing HubExceptions:
// KeyNotFoundException -> notFoundMessage, InvalidOperationException -> its own message.
private static async Task HubGuard(Func<Task> action, string notFoundMessage = "task not found")
src/ClaudeDo.Worker/Lifecycle/RunningOperationRegistry.cs
+49
View File
@@ -0,0 +1,49 @@
using System.Collections.Concurrent;
using ClaudeDo.Worker.Hub;
namespace ClaudeDo.Worker.Lifecycle;
/// <summary>
/// In-memory registry of Worker operations running right now (verify, merge, rebase, worktree
/// creation). <see cref="Begin"/> registers and broadcasts; disposing the returned scope removes
/// the entry, so a failed or thrown operation is always cleaned up by its caller's using/finally.
/// </summary>
public sealed class RunningOperationRegistry
{
private readonly ConcurrentDictionary<string, RunningOperationDto> _active = new();
private readonly HubBroadcaster? _broadcaster;
public RunningOperationRegistry(HubBroadcaster? broadcaster = null) => _broadcaster = broadcaster;
/// <summary>The operations running right now, oldest first.</summary>
public IReadOnlyList<RunningOperationDto> Snapshot() =>
_active.Values.OrderBy(o => o.StartedAt).ToList();
/// <summary>Registers an operation; dispose the result when it ends (idempotent).</summary>
public IDisposable Begin(string kind, string title, string? taskId = null, int? taskNumber = null, string? listId = null)
{
var op = new RunningOperationDto(Guid.NewGuid().ToString("N"), kind, taskId, taskNumber, listId, title, DateTime.UtcNow);
_active[op.Id] = op;
_ = Broadcast(_broadcaster?.OperationStarted(op));
return new Scope(this, op.Id);
}
// A broadcast failure must never fail the operation it describes.
private static async Task Broadcast(Task? send)
{
if (send is null) return;
try { await send; } catch { /* no clients / hub down */ }
}
private sealed class Scope(RunningOperationRegistry owner, string id) : IDisposable
{
private int _disposed;
public void Dispose()
{
if (Interlocked.Exchange(ref _disposed, 1) != 0) return;
owner._active.TryRemove(id, out _);
_ = Broadcast(owner._broadcaster?.OperationEnded(id));
}
}
}
src/ClaudeDo.Worker/Lifecycle/TaskMergeService.cs
+24 -9
View File
@@ -118,6 +118,7 @@ public sealed class TaskMergeService
private readonly IVerifyCommandRunner _verify;
private readonly ILogger<TaskMergeService> _logger;
private readonly UnitBranchService? _unitBranch;
private readonly RunningOperationRegistry _operations;
public TaskMergeService(
IDbContextFactory<ClaudeDoDbContext> dbFactory,
@@ -126,7 +127,8 @@ public sealed class TaskMergeService
ITaskStateService state,
IVerifyCommandRunner verify,
ILogger<TaskMergeService> logger,
UnitBranchService? unitBranch = null)
UnitBranchService? unitBranch = null,
RunningOperationRegistry? operations = null)
{
_dbFactory = dbFactory;
_git = git;
@@ -135,8 +137,12 @@ public sealed class TaskMergeService
_verify = verify;
_logger = logger;
_unitBranch = unitBranch;
_operations = operations ?? new RunningOperationRegistry();
}
private IDisposable TrackTask(string kind, TaskEntity task) =>
_operations.Begin(kind, task.Title, task.Id, task.Number, task.ListId);
private async Task<(TaskEntity Task, ListEntity List, WorktreeEntity? Worktree, string? VerifyCommand)> LoadMergeContextAsync(
string taskId, CancellationToken ct)
{
@@ -159,7 +165,7 @@ public sealed class TaskMergeService
/// CLAUDE.md — only the Done transition is withheld).
/// </summary>
private async Task<MergeResult?> RunVerifyGateAsync(
string? verifyCommand, string workingDir, CancellationToken ct,
string? verifyCommand, string workingDir, Func<IDisposable> track, CancellationToken ct,
IProgress<ProgressNotificationValue>? progress = null,
Action<TimeSpan>? onTick = null)
{
@@ -168,8 +174,9 @@ public sealed class TaskMergeService
VerifyCommandResult result;
try
{
result = await ProgressReporter.RunAsync(
_verify.RunAsync(workingDir, verifyCommand, VerifyTimeout, ct), ProgressReportInterval, progress, "verify gate running", onTick);
using (track())
result = await ProgressReporter.RunAsync(
_verify.RunAsync(workingDir, verifyCommand, VerifyTimeout, ct), ProgressReportInterval, progress, "verify gate running", onTick);
}
catch (Exception ex)
{
@@ -320,6 +327,8 @@ public sealed class TaskMergeService
}
if (candidates.Count == 0) return;
using var rebasing = TrackTask(OperationKinds.Rebasing, mergedTask);
// Runs where the stalled "Merging…" phase would otherwise sit unchanged in the UI while
// this loop does real (if best-effort) work — see the correction note in the task spec.
for (var i = 0; i < candidates.Count; i++)
@@ -399,6 +408,7 @@ public sealed class TaskMergeService
// Announced before the gate wait: another merge holding the repo is itself a reason the
// caller sees nothing happen, and a UI waiting on this call needs a phase to show at once.
using var merging = TrackTask(OperationKinds.Merging, task);
await _broadcaster.OperationProgress(taskId, PhaseMerging, 0, 0);
var gate = GetMergeGate(list.WorkingDir);
@@ -482,6 +492,7 @@ public sealed class TaskMergeService
}
await MarkWorktreeMergedAsync(taskId, mergeSha, ct);
merging.Dispose();
await RebaseOthersAfterMergeAsync(task, list, targetBranch, oldTargetTip, mergeSha, ct);
// The merge itself is instant; the verify gate is what makes this call take minutes.
@@ -506,7 +517,7 @@ public sealed class TaskMergeService
}
var verifyFailure = await RunVerifyGateAsync(
verifyCommand, list.WorkingDir, ct, progress,
verifyCommand, list.WorkingDir, () => TrackTask(OperationKinds.VerifyMerge, task), ct, progress,
elapsed => _ = _broadcaster.OperationProgress(taskId, PhaseVerifying, (int)elapsed.TotalSeconds, 0));
if (verifyFailure is not null)
{
@@ -641,7 +652,8 @@ public sealed class TaskMergeService
return new MergeResult(StatusMergedVerifyPending, Array.Empty<string>(), null);
}
var verifyFailure = await RunVerifyGateAsync(verifyCommand, list.WorkingDir, ct, progress);
var verifyFailure = await RunVerifyGateAsync(
verifyCommand, list.WorkingDir, () => TrackTask(OperationKinds.VerifyMerge, task), ct, progress);
if (verifyFailure is not null)
{
_logger.LogWarning("Verify command failed after continuing merge of task {TaskId}: {Reason}", taskId, verifyFailure.ErrorMessage);
@@ -1101,7 +1113,7 @@ public sealed class TaskMergeService
{
// Same reason as the post-merge gate: this holds the approve call for minutes.
var failed = await RunVerifyGateAsync(
verifyCommand, list.WorkingDir!, ct, progress,
verifyCommand, list.WorkingDir!, () => TrackTask(OperationKinds.VerifyMerge, task), ct, progress,
elapsed => _ = _broadcaster.OperationProgress(taskId, PhaseVerifying, (int)elapsed.TotalSeconds, 0));
if (failed is not null) return failed;
}
@@ -1141,6 +1153,7 @@ public sealed class TaskMergeService
{
var taskId = task.Id;
var workingDir = list.WorkingDir!;
using var merging = TrackTask(OperationKinds.Merging, task);
await _broadcaster.OperationProgress(taskId, PhaseMerging, 0, 0);
var gate = GetMergeGate(workingDir);
@@ -1195,6 +1208,7 @@ public sealed class TaskMergeService
}
await MarkWorktreeMergedAsync(taskId, mergeSha, ct);
merging.Dispose();
await RebaseOthersAfterMergeAsync(task, list, targetBranch, oldTargetTip, mergeSha, ct);
if (!string.IsNullOrWhiteSpace(verifyCommand) && skipVerify)
@@ -1206,7 +1220,7 @@ public sealed class TaskMergeService
}
var verifyFailure = await RunVerifyGateAsync(
verifyCommand, workingDir, ct, progress,
verifyCommand, workingDir, () => TrackTask(OperationKinds.VerifyMerge, task), ct, progress,
elapsed => _ = _broadcaster.OperationProgress(taskId, PhaseVerifying, (int)elapsed.TotalSeconds, 0));
if (verifyFailure is not null)
{
@@ -1284,7 +1298,8 @@ public sealed class TaskMergeService
$"Batch verify running for {candidates.Count} merged task(s) in \"{list.Name}\"",
WorkerLogLevel.Info, DateTime.UtcNow);
failure = await RunVerifyGateAsync(
verifyCommand, list.WorkingDir, ct, progress,
verifyCommand, list.WorkingDir,
() => _operations.Begin(OperationKinds.VerifyBatch, list.Name, listId: listId), ct, progress,
elapsed => _ = _broadcaster.OperationProgress(listId, PhaseVerifying, (int)elapsed.TotalSeconds, 0));
}
finally { gate.Release(); }
src/ClaudeDo.Worker/Program.cs
+1
View File
@@ -110,6 +110,7 @@ builder.Services.AddSignalR(options =>
builder.Services.AddSingleton<IClaudeProcess, ClaudeProcess>();
builder.Services.AddSingleton<HubBroadcaster>();
builder.Services.AddSingleton<ConnectedClientTracker>();
builder.Services.AddSingleton<RunningOperationRegistry>();
builder.Services.AddSingleton<GitService>();
builder.Services.AddSingleton<WorktreeManager>();
builder.Services.AddSingleton<UnitBranchService>();
src/ClaudeDo.Worker/Runner/WorktreeManager.cs
+6 -1
View File
@@ -5,6 +5,7 @@ using ClaudeDo.Data.Repositories;
using ClaudeDo.Worker.Config;
using ClaudeDo.Worker.Git;
using ClaudeDo.Worker.Hub;
using ClaudeDo.Worker.Lifecycle;
using ClaudeDo.Worker.Tickets;
using Microsoft.EntityFrameworkCore;
@@ -24,16 +25,19 @@ public sealed class WorktreeManager
private readonly WorkerConfig _cfg;
private readonly ILogger<WorktreeManager> _logger;
private readonly HubBroadcaster? _broadcaster;
private readonly RunningOperationRegistry _operations;
public WorktreeManager(
GitService git, IDbContextFactory<ClaudeDoDbContext> dbFactory, WorkerConfig cfg,
ILogger<WorktreeManager> logger, HubBroadcaster? broadcaster = null)
ILogger<WorktreeManager> logger, HubBroadcaster? broadcaster = null,
RunningOperationRegistry? operations = null)
{
_git = git;
_dbFactory = dbFactory;
_cfg = cfg;
_logger = logger;
_broadcaster = broadcaster;
_operations = operations ?? new RunningOperationRegistry();
}
private Task BroadcastCreating(string taskId) =>
@@ -115,6 +119,7 @@ public sealed class WorktreeManager
// Create the worktree. If a stale branch from a previous run remains
// (e.g. after force-remove), delete it and retry once.
using var creating = _operations.Begin(OperationKinds.CreatingWorktree, task.Title, task.Id, task.Number, task.ListId);
await BroadcastCreating(task.Id);
try
{