feat(claude-do): merge Show currently running Worker operations (verify, merge, reb
ClaudeDo-Task: ea44c9dd0c4942bfa3490bc3c1014205
This commit is contained in:
@@ -13,6 +13,21 @@ namespace ClaudeDo.Data.Wire;
|
||||
|
||||
public record ActiveTaskDto(string Slot, string TaskId, DateTime StartedAt);
|
||||
|
||||
/// <summary>A Worker operation running right now (verify, merge, rebase, worktree creation).
|
||||
/// <paramref name="Title"/> is the task title, or the list name for a batch verify.</summary>
|
||||
public record RunningOperationDto(
|
||||
string Id, string Kind, string? TaskId, int? TaskNumber, string? ListId, string Title, DateTime StartedAt);
|
||||
|
||||
/// <summary>Stable <see cref="RunningOperationDto.Kind"/> tokens, localized by the UI.</summary>
|
||||
public static class OperationKinds
|
||||
{
|
||||
public const string VerifyMerge = "verify-merge";
|
||||
public const string VerifyBatch = "verify-batch";
|
||||
public const string Merging = "merging";
|
||||
public const string Rebasing = "rebasing";
|
||||
public const string CreatingWorktree = "creating-worktree";
|
||||
}
|
||||
|
||||
public record AppSettingsDto(
|
||||
string DefaultClaudeInstructions,
|
||||
string DefaultModel,
|
||||
|
||||
@@ -375,6 +375,13 @@
|
||||
"empty": "Keine laufenden Aufgaben",
|
||||
"settings": "Einstellungen",
|
||||
"activity": "Aktivität",
|
||||
"runningNow": "Läuft gerade",
|
||||
"opVerifyMerge": "Verify (Merge)",
|
||||
"opVerifyBatch": "Verify (Batch)",
|
||||
"opMerging": "Merge",
|
||||
"opRebasing": "Rebase",
|
||||
"opCreatingWorktree": "Worktree wird erstellt",
|
||||
"operationsRefreshFailed": "Laufende Operationen konnten nicht geladen werden: {0}",
|
||||
"blocked": "Blockiert",
|
||||
"running": "Läuft",
|
||||
"review": "Im Review",
|
||||
|
||||
@@ -375,6 +375,13 @@
|
||||
"empty": "No running tasks",
|
||||
"settings": "Settings",
|
||||
"activity": "Activity",
|
||||
"runningNow": "Running now",
|
||||
"opVerifyMerge": "Verify (merge)",
|
||||
"opVerifyBatch": "Verify (batch)",
|
||||
"opMerging": "Merging",
|
||||
"opRebasing": "Rebasing",
|
||||
"opCreatingWorktree": "Creating worktree",
|
||||
"operationsRefreshFailed": "Couldn't load running operations: {0}",
|
||||
"blocked": "Blocked",
|
||||
"running": "Running",
|
||||
"review": "In review",
|
||||
|
||||
@@ -59,6 +59,11 @@ public interface IWorkerClient : INotifyPropertyChanged
|
||||
/// total.</summary>
|
||||
event Action<string, string, int, int>? OperationProgressEvent;
|
||||
|
||||
/// <summary>A Worker operation (verify, merge, rebase, worktree creation) began.</summary>
|
||||
event Action<RunningOperationDto>? OperationStartedEvent;
|
||||
/// <summary>The operation with this id finished, failed or threw.</summary>
|
||||
event Action<string>? OperationEndedEvent;
|
||||
|
||||
event Action<string, string>? PlanningMergeStartedEvent;
|
||||
event Action<string, string>? PlanningSubtaskMergedEvent;
|
||||
/// <summary>(planningTaskId, subtaskId, conflictedFiles, externallyDriven). externallyDriven
|
||||
@@ -208,6 +213,9 @@ public interface IWorkerClient : INotifyPropertyChanged
|
||||
Task<string> GetLastPrepLogAsync();
|
||||
Task<IReadOnlyList<WorkerLogEntry>> GetRecentLogsAsync();
|
||||
|
||||
/// <summary>Operations running on the Worker right now — the state to start from before applying started/ended events.</summary>
|
||||
Task<IReadOnlyList<RunningOperationDto>> GetActiveOperationsAsync();
|
||||
|
||||
Task<List<PrimeScheduleDto>> GetPrimeSchedulesAsync();
|
||||
Task<PrimeScheduleDto?> UpsertPrimeScheduleAsync(PrimeScheduleDto dto);
|
||||
Task DeletePrimeScheduleAsync(Guid id);
|
||||
|
||||
@@ -69,6 +69,8 @@ public partial class WorkerClient : ObservableObject, IAsyncDisposable, IWorkerC
|
||||
public event Action<UsageSnapshotDto>? UsageUpdatedEvent;
|
||||
|
||||
public event Action<string, string, int, int>? OperationProgressEvent;
|
||||
public event Action<RunningOperationDto>? OperationStartedEvent;
|
||||
public event Action<string>? OperationEndedEvent;
|
||||
|
||||
public event Action<string, string>? PlanningMergeStartedEvent;
|
||||
public event Action<string, string>? PlanningSubtaskMergedEvent;
|
||||
@@ -203,6 +205,16 @@ public partial class WorkerClient : ObservableObject, IAsyncDisposable, IWorkerC
|
||||
Dispatcher.UIThread.Post(() => OperationProgressEvent?.Invoke(opKey, phase, current, total));
|
||||
});
|
||||
|
||||
_hub.On<RunningOperationDto>("OperationStarted", op =>
|
||||
{
|
||||
Dispatcher.UIThread.Post(() => OperationStartedEvent?.Invoke(op));
|
||||
});
|
||||
|
||||
_hub.On<string>("OperationEnded", id =>
|
||||
{
|
||||
Dispatcher.UIThread.Post(() => OperationEndedEvent?.Invoke(id));
|
||||
});
|
||||
|
||||
_hub.On<string, string>("PlanningMergeStarted", (planningTaskId, targetBranch) =>
|
||||
{
|
||||
Dispatcher.UIThread.Post(() => PlanningMergeStartedEvent?.Invoke(planningTaskId, targetBranch));
|
||||
@@ -530,6 +542,9 @@ public partial class WorkerClient : ObservableObject, IAsyncDisposable, IWorkerC
|
||||
public async Task<string> GetLastPrepLogAsync()
|
||||
=> await TryInvokeAsync<string>("GetLastPrepLog") ?? string.Empty;
|
||||
|
||||
public async Task<IReadOnlyList<RunningOperationDto>> GetActiveOperationsAsync()
|
||||
=> await TryInvokeAsync<List<RunningOperationDto>>("GetActiveOperations") ?? new List<RunningOperationDto>();
|
||||
|
||||
public async Task<IReadOnlyList<WorkerLogEntry>> GetRecentLogsAsync()
|
||||
=> await TryInvokeAsync<List<WorkerLogEntry>>("GetRecentLogs") ?? new List<WorkerLogEntry>();
|
||||
|
||||
|
||||
@@ -23,6 +23,8 @@ public sealed partial class MissionControlViewModel : ViewModelBase, IDisposable
|
||||
private readonly Action<string, string, string, DateTime> _onTaskFinished;
|
||||
private readonly Action<string> _onTaskUpdated;
|
||||
private readonly Action _onConnectionRestored;
|
||||
private readonly Action<RunningOperationDto> _onOperationStarted;
|
||||
private readonly Action<string> _onOperationEnded;
|
||||
private readonly Action<string, IReadOnlyList<string>, string> _onHandoffRequested;
|
||||
private readonly Action<string, IReadOnlyList<string>> _onListHandlerStartRequested;
|
||||
private readonly Action<string, string> _onHandlerUsageWarning;
|
||||
@@ -78,6 +80,13 @@ public sealed partial class MissionControlViewModel : ViewModelBase, IDisposable
|
||||
public ObservableCollection<QueuedTaskViewModel> Queued { get; } = new();
|
||||
public bool HasQueued => Queued.Count > 0;
|
||||
|
||||
/// <summary>Worker operations running right now (verify, merge, rebase, worktree creation), oldest first.</summary>
|
||||
public ObservableCollection<RunningOperationViewModel> RunningOperations { get; } = new();
|
||||
public bool HasRunningOperations => RunningOperations.Count > 0;
|
||||
|
||||
/// <summary>The activity strip is shown while it has queue rows or running operations.</summary>
|
||||
public bool HasActivity => HasQueued || HasRunningOperations;
|
||||
|
||||
// "New session" targets: one entry per list with a working dir, plus a browse fallback, so a
|
||||
// fresh session is two clicks instead of navigating a folder picker.
|
||||
public ObservableCollection<SessionTargetViewModel> SessionTargets { get; } = new();
|
||||
@@ -118,9 +127,14 @@ public sealed partial class MissionControlViewModel : ViewModelBase, IDisposable
|
||||
_onTaskUpdated = taskId => _ = RefreshQueueAsync();
|
||||
_worker.TaskUpdatedEvent += _onTaskUpdated;
|
||||
|
||||
_onConnectionRestored = () => { _ = RefreshQueueAsync(); };
|
||||
_onConnectionRestored = () => { _ = RefreshQueueAsync(); _ = RefreshOperationsAsync(); };
|
||||
_worker.ConnectionRestoredEvent += _onConnectionRestored;
|
||||
|
||||
_onOperationStarted = AddOperation;
|
||||
_worker.OperationStartedEvent += _onOperationStarted;
|
||||
_onOperationEnded = RemoveOperation;
|
||||
_worker.OperationEndedEvent += _onOperationEnded;
|
||||
|
||||
_onHandoffRequested = (taskId, survivingTaskIds, nextPhase) => { _ = OpenMergeHelperHandoffConPtySessionAsync(taskId, survivingTaskIds, nextPhase); };
|
||||
_worker.HandoffRequestedEvent += _onHandoffRequested;
|
||||
_worker.SessionHandoffRequestedEvent += OpenSessionHandoff;
|
||||
@@ -133,9 +147,52 @@ public sealed partial class MissionControlViewModel : ViewModelBase, IDisposable
|
||||
_worker.HandlerHandoffDeferredEvent += _onHandlerHandoffDeferred;
|
||||
|
||||
_ = RefreshQueueAsync();
|
||||
_ = RefreshOperationsAsync();
|
||||
_ = RefreshSessionTargetsAsync();
|
||||
}
|
||||
|
||||
/// <summary>Replaces the running-operations list with the Worker's snapshot (open and reconnect).</summary>
|
||||
internal async System.Threading.Tasks.Task RefreshOperationsAsync()
|
||||
{
|
||||
try
|
||||
{
|
||||
var snapshot = await _worker.GetActiveOperationsAsync();
|
||||
RunningOperations.Clear();
|
||||
foreach (var op in snapshot.OrderBy(o => o.StartedAt))
|
||||
RunningOperations.Add(new RunningOperationViewModel(op));
|
||||
NotifyOperationsChanged();
|
||||
}
|
||||
catch (Exception ex) { ErrorReported?.Invoke(Loc.T("missionControl.operationsRefreshFailed", ex.Message)); }
|
||||
}
|
||||
|
||||
private void AddOperation(RunningOperationDto op)
|
||||
{
|
||||
if (RunningOperations.Any(r => r.Id == op.Id)) return;
|
||||
RunningOperations.Add(new RunningOperationViewModel(op));
|
||||
NotifyOperationsChanged();
|
||||
}
|
||||
|
||||
private void RemoveOperation(string id)
|
||||
{
|
||||
var row = RunningOperations.FirstOrDefault(r => r.Id == id);
|
||||
if (row is null) return;
|
||||
RunningOperations.Remove(row);
|
||||
NotifyOperationsChanged();
|
||||
}
|
||||
|
||||
private void NotifyOperationsChanged()
|
||||
{
|
||||
OnPropertyChanged(nameof(HasRunningOperations));
|
||||
OnPropertyChanged(nameof(HasActivity));
|
||||
}
|
||||
|
||||
/// <summary>Refreshes every row's elapsed text; the view calls this once a second while open.</summary>
|
||||
public void TickOperations(DateTime? nowUtc = null)
|
||||
{
|
||||
var now = nowUtc ?? DateTime.UtcNow;
|
||||
foreach (var r in RunningOperations) r.Tick(now);
|
||||
}
|
||||
|
||||
internal async System.Threading.Tasks.Task RefreshQueueAsync()
|
||||
{
|
||||
try
|
||||
@@ -174,6 +231,7 @@ public sealed partial class MissionControlViewModel : ViewModelBase, IDisposable
|
||||
});
|
||||
}
|
||||
OnPropertyChanged(nameof(HasQueued));
|
||||
OnPropertyChanged(nameof(HasActivity));
|
||||
}
|
||||
catch (Exception ex) { ErrorReported?.Invoke(Loc.T("missionControl.queueRefreshFailed", ex.Message)); }
|
||||
}
|
||||
@@ -777,6 +835,8 @@ public sealed partial class MissionControlViewModel : ViewModelBase, IDisposable
|
||||
_worker.TaskFinishedEvent -= _onTaskFinished;
|
||||
_worker.TaskUpdatedEvent -= _onTaskUpdated;
|
||||
_worker.ConnectionRestoredEvent -= _onConnectionRestored;
|
||||
_worker.OperationStartedEvent -= _onOperationStarted;
|
||||
_worker.OperationEndedEvent -= _onOperationEnded;
|
||||
_worker.HandoffRequestedEvent -= _onHandoffRequested;
|
||||
_worker.SessionHandoffRequestedEvent -= OpenSessionHandoff;
|
||||
_worker.ListHandlerStartRequestedEvent -= _onListHandlerStartRequested;
|
||||
@@ -820,6 +880,45 @@ public sealed class QueuedTaskViewModel
|
||||
public IRelayCommand? OpenSessionCommand { get; init; }
|
||||
}
|
||||
|
||||
/// <summary>One row in the "Running now" section: a Worker operation with a ticking elapsed time.</summary>
|
||||
public sealed partial class RunningOperationViewModel : ObservableObject
|
||||
{
|
||||
private readonly DateTime _startedAtUtc;
|
||||
|
||||
public string Id { get; }
|
||||
/// <summary>Localized operation kind ("Verify (merge)", "Rebasing", ...).</summary>
|
||||
public string KindLabel { get; }
|
||||
/// <summary><c>#Number Title</c> for a task operation, the list name for a batch verify.</summary>
|
||||
public string Subject { get; }
|
||||
|
||||
[ObservableProperty] private string _elapsed;
|
||||
|
||||
public RunningOperationViewModel(RunningOperationDto op)
|
||||
{
|
||||
Id = op.Id;
|
||||
_startedAtUtc = op.StartedAt;
|
||||
KindLabel = op.Kind switch
|
||||
{
|
||||
OperationKinds.VerifyMerge => Loc.T("missionControl.opVerifyMerge"),
|
||||
OperationKinds.VerifyBatch => Loc.T("missionControl.opVerifyBatch"),
|
||||
OperationKinds.Merging => Loc.T("missionControl.opMerging"),
|
||||
OperationKinds.Rebasing => Loc.T("missionControl.opRebasing"),
|
||||
OperationKinds.CreatingWorktree => Loc.T("missionControl.opCreatingWorktree"),
|
||||
_ => op.Kind,
|
||||
};
|
||||
Subject = op.TaskNumber is > 0 ? $"#{op.TaskNumber} {op.Title}" : op.Title;
|
||||
_elapsed = Format(DateTime.UtcNow - _startedAtUtc);
|
||||
}
|
||||
|
||||
public void Tick(DateTime nowUtc) => Elapsed = Format(nowUtc - _startedAtUtc);
|
||||
|
||||
private static string Format(TimeSpan t)
|
||||
{
|
||||
if (t < TimeSpan.Zero) t = TimeSpan.Zero;
|
||||
return t.TotalHours >= 1 ? t.ToString(@"h\:mm\:ss") : t.ToString(@"mm\:ss");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>One row in the "Recent sessions" flyout.</summary>
|
||||
public sealed class RecentSessionViewModel
|
||||
{
|
||||
|
||||
@@ -185,7 +185,7 @@
|
||||
</Panel>
|
||||
|
||||
<GridSplitter x:Name="QueueSplitter" Grid.Column="1"
|
||||
IsVisible="{Binding HasQueued}"
|
||||
IsVisible="{Binding HasActivity}"
|
||||
Width="6"
|
||||
Background="Transparent"
|
||||
HorizontalAlignment="Stretch" VerticalAlignment="Stretch"
|
||||
@@ -194,12 +194,49 @@
|
||||
|
||||
<!-- Read-only queue strip — collapses when nothing is queued -->
|
||||
<Border Grid.Column="2"
|
||||
IsVisible="{Binding HasQueued}"
|
||||
IsVisible="{Binding HasActivity}"
|
||||
Background="{DynamicResource DeepBrush}"
|
||||
BorderBrush="{DynamicResource LineBrush}"
|
||||
BorderThickness="1,0,0,0">
|
||||
<DockPanel LastChildFill="True" Margin="10,10">
|
||||
<!-- Worker operations running right now; hidden when nothing runs. -->
|
||||
<StackPanel DockPanel.Dock="Top" Margin="0,0,0,12" Spacing="4"
|
||||
IsVisible="{Binding HasRunningOperations}">
|
||||
<TextBlock Classes="eyebrow"
|
||||
Text="{loc:Tr missionControl.runningNow}"
|
||||
Foreground="{DynamicResource TextMuteBrush}"
|
||||
LetterSpacing="1.4" Margin="0,0,0,4" />
|
||||
<ItemsControl ItemsSource="{Binding RunningOperations}">
|
||||
<ItemsControl.ItemTemplate>
|
||||
<DataTemplate x:DataType="vm:RunningOperationViewModel">
|
||||
<Panel Margin="0,0,0,4">
|
||||
<Border Background="{DynamicResource SurfaceBrush}"
|
||||
BorderBrush="{DynamicResource LineBrush}"
|
||||
BorderThickness="1" CornerRadius="6" />
|
||||
<Border Background="{DynamicResource RunningTintBrush}"
|
||||
BorderBrush="{DynamicResource RunningTintBorderBrush}"
|
||||
BorderThickness="1" CornerRadius="6" />
|
||||
<StackPanel Margin="8,6" Spacing="2">
|
||||
<TextBlock Text="{Binding Subject}"
|
||||
TextTrimming="CharacterEllipsis"
|
||||
ToolTip.Tip="{Binding Subject}"
|
||||
Foreground="{DynamicResource TextDimBrush}" />
|
||||
<Grid ColumnDefinitions="*,Auto">
|
||||
<TextBlock Grid.Column="0" Classes="meta"
|
||||
Text="{Binding KindLabel}"
|
||||
Foreground="{DynamicResource StatusRunningBrush}" />
|
||||
<TextBlock Grid.Column="1" Classes="meta"
|
||||
Text="{Binding Elapsed}"
|
||||
Foreground="{DynamicResource TextMuteBrush}" />
|
||||
</Grid>
|
||||
</StackPanel>
|
||||
</Panel>
|
||||
</DataTemplate>
|
||||
</ItemsControl.ItemTemplate>
|
||||
</ItemsControl>
|
||||
</StackPanel>
|
||||
<TextBlock DockPanel.Dock="Top" Classes="eyebrow"
|
||||
IsVisible="{Binding HasQueued}"
|
||||
Text="{loc:Tr missionControl.activity}"
|
||||
Foreground="{DynamicResource TextMuteBrush}"
|
||||
LetterSpacing="1.4" Margin="0,0,0,8" />
|
||||
|
||||
@@ -40,6 +40,24 @@ public partial class MissionControlView : UserControl
|
||||
OverviewGrid.AddHandler(PointerReleasedEvent, OnTilePointerReleased);
|
||||
}
|
||||
|
||||
// Ticks the "Running now" elapsed times while the view is on screen.
|
||||
private Avalonia.Threading.DispatcherTimer? _elapsedTimer;
|
||||
|
||||
protected override void OnAttachedToVisualTree(VisualTreeAttachmentEventArgs e)
|
||||
{
|
||||
base.OnAttachedToVisualTree(e);
|
||||
_elapsedTimer = new Avalonia.Threading.DispatcherTimer { Interval = TimeSpan.FromSeconds(1) };
|
||||
_elapsedTimer.Tick += (_, _) => _vm?.TickOperations();
|
||||
_elapsedTimer.Start();
|
||||
}
|
||||
|
||||
protected override void OnDetachedFromVisualTree(VisualTreeAttachmentEventArgs e)
|
||||
{
|
||||
_elapsedTimer?.Stop();
|
||||
_elapsedTimer = null;
|
||||
base.OnDetachedFromVisualTree(e);
|
||||
}
|
||||
|
||||
private const double DragThreshold = 4;
|
||||
|
||||
/// <summary>Tile presenter whose header was pressed; null when no tile drag is pending.</summary>
|
||||
@@ -199,7 +217,7 @@ public partial class MissionControlView : UserControl
|
||||
{
|
||||
if (e.PropertyName == nameof(MissionControlViewModel.ColumnCount))
|
||||
RebuildOverviewGrid();
|
||||
else if (e.PropertyName == nameof(MissionControlViewModel.HasQueued))
|
||||
else if (e.PropertyName == nameof(MissionControlViewModel.HasActivity))
|
||||
UpdateQueueColumn();
|
||||
else if (e.PropertyName == nameof(MissionControlViewModel.IsFocusMode))
|
||||
// The now-visible layout's hosts must re-steal their pane views from the hidden
|
||||
@@ -221,7 +239,7 @@ public partial class MissionControlView : UserControl
|
||||
private void UpdateQueueColumn()
|
||||
{
|
||||
var queueColumn = BodyGrid.ColumnDefinitions[2];
|
||||
if (_vm?.HasQueued == true)
|
||||
if (_vm?.HasActivity == true)
|
||||
{
|
||||
queueColumn.MinWidth = MinQueueWidth;
|
||||
queueColumn.Width = _queueWidth;
|
||||
|
||||
@@ -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);
|
||||
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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(); }
|
||||
|
||||
@@ -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>();
|
||||
|
||||
@@ -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
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user