feat(worker): add generic OperationProgress channel, port MergeProgress onto it
Merge/verify phases now broadcast over a generic (opKey, phase, current, total) Hub event instead of a merge-specific one, so future producers (worktree cleanup, startup recovery, planning integration) can reuse it. IWorkerClient.MergeProgressEvent stays as a thin forwarder for existing consumers (elapsed seconds riding in the generic "current" slot).
This commit is contained in:
@@ -34,9 +34,18 @@ public interface IWorkerClient : INotifyPropertyChanged
|
|||||||
event Action<string>? PrepLineEvent;
|
event Action<string>? PrepLineEvent;
|
||||||
event Action<bool>? PrepFinishedEvent;
|
event Action<bool>? PrepFinishedEvent;
|
||||||
|
|
||||||
|
/// <summary>(opKey, phase, current, total) — generic progress channel for long-running worker
|
||||||
|
/// operations (merge phases, worktree cleanup, startup recovery, planning integration, ...).
|
||||||
|
/// opKey is the TaskId for a task-bound operation, otherwise a stable string. current/total
|
||||||
|
/// is meant to be shown as text, never a progress bar — most operations have no meaningful
|
||||||
|
/// total.</summary>
|
||||||
|
event Action<string, string, int, int>? OperationProgressEvent;
|
||||||
|
|
||||||
/// <summary>(taskId, phase, elapsedSeconds) — phase of an in-flight single-task merge
|
/// <summary>(taskId, phase, elapsedSeconds) — phase of an in-flight single-task merge
|
||||||
/// ("merging" | "verifying"). Fires while the MergeTask call itself is still pending, so the
|
/// ("merging" | "verifying"). Fires while the MergeTask call itself is still pending, so the
|
||||||
/// waiting UI can show what it's blocked on; the verify phase re-fires every 30 s.</summary>
|
/// waiting UI can show what it's blocked on; the verify phase re-fires every 30 s. A thin
|
||||||
|
/// forwarder over <see cref="OperationProgressEvent"/> kept for existing consumers (elapsed
|
||||||
|
/// seconds riding in the generic "current" slot).</summary>
|
||||||
event Action<string, string, int>? MergeProgressEvent;
|
event Action<string, string, int>? MergeProgressEvent;
|
||||||
|
|
||||||
event Action<string, string>? PlanningMergeStartedEvent;
|
event Action<string, string>? PlanningMergeStartedEvent;
|
||||||
|
|||||||
@@ -64,6 +64,7 @@ public partial class WorkerClient : ObservableObject, IAsyncDisposable, IWorkerC
|
|||||||
|
|
||||||
public event Action<UsageSnapshotDto>? UsageUpdatedEvent;
|
public event Action<UsageSnapshotDto>? UsageUpdatedEvent;
|
||||||
|
|
||||||
|
public event Action<string, string, int, int>? OperationProgressEvent;
|
||||||
public event Action<string, string, int>? MergeProgressEvent;
|
public event Action<string, string, int>? MergeProgressEvent;
|
||||||
|
|
||||||
public event Action<string, string>? PlanningMergeStartedEvent;
|
public event Action<string, string>? PlanningMergeStartedEvent;
|
||||||
@@ -174,9 +175,13 @@ public partial class WorkerClient : ObservableObject, IAsyncDisposable, IWorkerC
|
|||||||
WorkerLogReceivedEvent?.Invoke(new WorkerLogEntry(message, level, timestampUtc)));
|
WorkerLogReceivedEvent?.Invoke(new WorkerLogEntry(message, level, timestampUtc)));
|
||||||
});
|
});
|
||||||
|
|
||||||
_hub.On<string, string, int>("MergeProgress", (taskId, phase, elapsedSeconds) =>
|
_hub.On<string, string, int, int>("OperationProgress", (opKey, phase, current, total) =>
|
||||||
{
|
{
|
||||||
Dispatcher.UIThread.Post(() => MergeProgressEvent?.Invoke(taskId, phase, elapsedSeconds));
|
Dispatcher.UIThread.Post(() =>
|
||||||
|
{
|
||||||
|
OperationProgressEvent?.Invoke(opKey, phase, current, total);
|
||||||
|
MergeProgressEvent?.Invoke(opKey, phase, current);
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
_hub.On<string, string>("PlanningMergeStarted", (planningTaskId, targetBranch) =>
|
_hub.On<string, string>("PlanningMergeStarted", (planningTaskId, targetBranch) =>
|
||||||
|
|||||||
@@ -168,7 +168,9 @@ launch specs · worktrees · agents/settings/lists · reports/notes/prep · diag
|
|||||||
- `PrepStarted`
|
- `PrepStarted`
|
||||||
- `PrepLine`
|
- `PrepLine`
|
||||||
- `PrepFinished`
|
- `PrepFinished`
|
||||||
- `MergeProgress`
|
- `OperationProgress` (generic `(opKey, phase, current, total)` channel; merge phases are its
|
||||||
|
first producer — `IWorkerClient.MergeProgressEvent` on the Ui side is a thin forwarder over it
|
||||||
|
for existing consumers, elapsed seconds riding in the `current` slot)
|
||||||
- `PlanningMergeStarted`
|
- `PlanningMergeStarted`
|
||||||
- `PlanningSubtaskMerged`
|
- `PlanningSubtaskMerged`
|
||||||
- `PlanningMergeConflict`
|
- `PlanningMergeConflict`
|
||||||
|
|||||||
@@ -46,10 +46,13 @@ public sealed class HubBroadcaster : IPrimeBroadcaster, IRefineBroadcaster
|
|||||||
public Task WorkerLog(string message, WorkerLogLevel level, DateTime timestampUtc) =>
|
public Task WorkerLog(string message, WorkerLogLevel level, DateTime timestampUtc) =>
|
||||||
_hub.Clients.All.SendAsync("WorkerLog", message, level, timestampUtc);
|
_hub.Clients.All.SendAsync("WorkerLog", message, level, timestampUtc);
|
||||||
|
|
||||||
// Phase of an in-flight single-task merge (see TaskMergeService.Phase*), so a client waiting
|
// Generic progress channel for long-running worker operations (merge phases, worktree
|
||||||
// on the MergeTask call can show what it is waiting for instead of a frozen button.
|
// cleanup, startup recovery, planning integration, ...). opKey is the TaskId for a
|
||||||
public Task MergeProgress(string taskId, string phase, int elapsedSeconds) =>
|
// task-bound operation, otherwise a stable string ("worktree-cleanup", "startup-recovery",
|
||||||
_hub.Clients.All.SendAsync("MergeProgress", taskId, phase, elapsedSeconds);
|
// "planning-integration:<taskId>"). current/total is shown as text by the UI, never a
|
||||||
|
// progress bar -- most operations have no meaningful total.
|
||||||
|
public Task OperationProgress(string opKey, string phase, int current, int total) =>
|
||||||
|
_hub.Clients.All.SendAsync("OperationProgress", opKey, phase, current, total);
|
||||||
|
|
||||||
public Task PlanningMergeStarted(string planningTaskId, string targetBranch) =>
|
public Task PlanningMergeStarted(string planningTaskId, string targetBranch) =>
|
||||||
_hub.Clients.All.SendAsync("PlanningMergeStarted", planningTaskId, targetBranch);
|
_hub.Clients.All.SendAsync("PlanningMergeStarted", planningTaskId, targetBranch);
|
||||||
|
|||||||
@@ -61,7 +61,7 @@ public sealed class TaskMergeService
|
|||||||
public const string StatusReverted = "reverted";
|
public const string StatusReverted = "reverted";
|
||||||
public const string StatusConflictAborted = "conflict_aborted";
|
public const string StatusConflictAborted = "conflict_aborted";
|
||||||
|
|
||||||
// Phase tokens for the MergeProgress broadcast — stable identifiers, localized by the UI.
|
// Phase tokens for the OperationProgress broadcast — stable identifiers, localized by the UI.
|
||||||
public const string PhaseMerging = "merging";
|
public const string PhaseMerging = "merging";
|
||||||
public const string PhaseVerifying = "verifying";
|
public const string PhaseVerifying = "verifying";
|
||||||
|
|
||||||
@@ -168,7 +168,7 @@ public sealed class TaskMergeService
|
|||||||
/// Awaits <paramref name="work"/> while reporting MCP progress every
|
/// Awaits <paramref name="work"/> while reporting MCP progress every
|
||||||
/// <see cref="ProgressReportInterval"/> so a caller waiting on a long verify run doesn't hit
|
/// <see cref="ProgressReportInterval"/> so a caller waiting on a long verify run doesn't hit
|
||||||
/// the MCP client's own idle-silence abort. <paramref name="onTick"/> rides the same cadence
|
/// the MCP client's own idle-silence abort. <paramref name="onTick"/> rides the same cadence
|
||||||
/// for non-MCP callers (the Hub, which turns it into a MergeProgress broadcast). No-op
|
/// for non-MCP callers (the Hub, which turns it into an OperationProgress broadcast). No-op
|
||||||
/// passthrough when both are null.
|
/// passthrough when both are null.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
private static async Task<T> RunReportingProgressAsync<T>(
|
private static async Task<T> RunReportingProgressAsync<T>(
|
||||||
@@ -392,7 +392,7 @@ public sealed class TaskMergeService
|
|||||||
|
|
||||||
// Announced before the gate wait: another merge holding the repo is itself a reason the
|
// 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.
|
// caller sees nothing happen, and a UI waiting on this call needs a phase to show at once.
|
||||||
await _broadcaster.MergeProgress(taskId, PhaseMerging, 0);
|
await _broadcaster.OperationProgress(taskId, PhaseMerging, 0, 0);
|
||||||
|
|
||||||
var gate = GetMergeGate(list.WorkingDir);
|
var gate = GetMergeGate(list.WorkingDir);
|
||||||
await gate.WaitAsync(ct);
|
await gate.WaitAsync(ct);
|
||||||
@@ -479,7 +479,7 @@ public sealed class TaskMergeService
|
|||||||
// silence here is what makes a working merge look like a dead button.
|
// silence here is what makes a working merge look like a dead button.
|
||||||
if (!string.IsNullOrWhiteSpace(verifyCommand))
|
if (!string.IsNullOrWhiteSpace(verifyCommand))
|
||||||
{
|
{
|
||||||
await _broadcaster.MergeProgress(taskId, PhaseVerifying, 0);
|
await _broadcaster.OperationProgress(taskId, PhaseVerifying, 0, 0);
|
||||||
await _broadcaster.WorkerLog(
|
await _broadcaster.WorkerLog(
|
||||||
$"Verify command running after merging #{task.Number} \"{task.Title}\" into {targetBranch}",
|
$"Verify command running after merging #{task.Number} \"{task.Title}\" into {targetBranch}",
|
||||||
WorkerLogLevel.Info, DateTime.UtcNow);
|
WorkerLogLevel.Info, DateTime.UtcNow);
|
||||||
@@ -487,7 +487,7 @@ public sealed class TaskMergeService
|
|||||||
|
|
||||||
var verifyFailure = await RunVerifyGateAsync(
|
var verifyFailure = await RunVerifyGateAsync(
|
||||||
verifyCommand, list.WorkingDir, ct, progress,
|
verifyCommand, list.WorkingDir, ct, progress,
|
||||||
elapsed => _ = _broadcaster.MergeProgress(taskId, PhaseVerifying, (int)elapsed.TotalSeconds));
|
elapsed => _ = _broadcaster.OperationProgress(taskId, PhaseVerifying, (int)elapsed.TotalSeconds, 0));
|
||||||
if (verifyFailure is not null)
|
if (verifyFailure is not null)
|
||||||
{
|
{
|
||||||
_logger.LogWarning("Verify command failed after merging task {TaskId}: {Reason}", taskId, verifyFailure.ErrorMessage);
|
_logger.LogWarning("Verify command failed after merging task {TaskId}: {Reason}", taskId, verifyFailure.ErrorMessage);
|
||||||
@@ -897,14 +897,14 @@ public sealed class TaskMergeService
|
|||||||
if (!string.IsNullOrWhiteSpace(verifyCommand) && !string.IsNullOrWhiteSpace(list.WorkingDir))
|
if (!string.IsNullOrWhiteSpace(verifyCommand) && !string.IsNullOrWhiteSpace(list.WorkingDir))
|
||||||
{
|
{
|
||||||
var verifyGate = GetMergeGate(list.WorkingDir!);
|
var verifyGate = GetMergeGate(list.WorkingDir!);
|
||||||
await _broadcaster.MergeProgress(taskId, PhaseVerifying, 0);
|
await _broadcaster.OperationProgress(taskId, PhaseVerifying, 0, 0);
|
||||||
await verifyGate.WaitAsync(ct);
|
await verifyGate.WaitAsync(ct);
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
// Same reason as the post-merge gate: this holds the approve call for minutes.
|
// Same reason as the post-merge gate: this holds the approve call for minutes.
|
||||||
var failed = await RunVerifyGateAsync(
|
var failed = await RunVerifyGateAsync(
|
||||||
verifyCommand, list.WorkingDir!, ct, progress,
|
verifyCommand, list.WorkingDir!, ct, progress,
|
||||||
elapsed => _ = _broadcaster.MergeProgress(taskId, PhaseVerifying, (int)elapsed.TotalSeconds));
|
elapsed => _ = _broadcaster.OperationProgress(taskId, PhaseVerifying, (int)elapsed.TotalSeconds, 0));
|
||||||
if (failed is not null) return failed;
|
if (failed is not null) return failed;
|
||||||
}
|
}
|
||||||
finally { verifyGate.Release(); }
|
finally { verifyGate.Release(); }
|
||||||
|
|||||||
@@ -31,6 +31,7 @@ public abstract class StubWorkerClient : IWorkerClient
|
|||||||
public event Action<bool>? PrepFinishedEvent;
|
public event Action<bool>? PrepFinishedEvent;
|
||||||
public event Action<string>? RefineStartedEvent;
|
public event Action<string>? RefineStartedEvent;
|
||||||
public event Action<string, bool, string?>? RefineFinishedEvent;
|
public event Action<string, bool, string?>? RefineFinishedEvent;
|
||||||
|
public event Action<string, string, int, int>? OperationProgressEvent;
|
||||||
public event Action<string, string, int>? MergeProgressEvent;
|
public event Action<string, string, int>? MergeProgressEvent;
|
||||||
public event Action<string, string>? PlanningMergeStartedEvent;
|
public event Action<string, string>? PlanningMergeStartedEvent;
|
||||||
public event Action<string, string>? PlanningSubtaskMergedEvent;
|
public event Action<string, string>? PlanningSubtaskMergedEvent;
|
||||||
@@ -56,6 +57,7 @@ public abstract class StubWorkerClient : IWorkerClient
|
|||||||
public void RaiseHandoffRequested(string taskId, IReadOnlyList<string> survivingTaskIds, string nextPhase = "wait") => HandoffRequestedEvent?.Invoke(taskId, survivingTaskIds, nextPhase);
|
public void RaiseHandoffRequested(string taskId, IReadOnlyList<string> survivingTaskIds, string nextPhase = "wait") => HandoffRequestedEvent?.Invoke(taskId, survivingTaskIds, nextPhase);
|
||||||
public void RaisePlanningMergeConflict(string planningTaskId, string subtaskId, IReadOnlyList<string> files, bool externallyDriven)
|
public void RaisePlanningMergeConflict(string planningTaskId, string subtaskId, IReadOnlyList<string> files, bool externallyDriven)
|
||||||
=> PlanningMergeConflictEvent?.Invoke(planningTaskId, subtaskId, files, externallyDriven);
|
=> PlanningMergeConflictEvent?.Invoke(planningTaskId, subtaskId, files, externallyDriven);
|
||||||
|
public void RaiseOperationProgress(string opKey, string phase, int current, int total) => OperationProgressEvent?.Invoke(opKey, phase, current, total);
|
||||||
public void RaiseMergeProgress(string taskId, string phase, int elapsedSeconds) => MergeProgressEvent?.Invoke(taskId, phase, elapsedSeconds);
|
public void RaiseMergeProgress(string taskId, string phase, int elapsedSeconds) => MergeProgressEvent?.Invoke(taskId, phase, elapsedSeconds);
|
||||||
public void RaisePlanningMergeStarted(string planningTaskId, string targetBranch) => PlanningMergeStartedEvent?.Invoke(planningTaskId, targetBranch);
|
public void RaisePlanningMergeStarted(string planningTaskId, string targetBranch) => PlanningMergeStartedEvent?.Invoke(planningTaskId, targetBranch);
|
||||||
public void RaisePlanningMergeAborted(string planningTaskId) => PlanningMergeAbortedEvent?.Invoke(planningTaskId);
|
public void RaisePlanningMergeAborted(string planningTaskId) => PlanningMergeAbortedEvent?.Invoke(planningTaskId);
|
||||||
|
|||||||
@@ -335,9 +335,9 @@ public class TaskMergeServiceTests : IDisposable
|
|||||||
commitMessage: "", ct: CancellationToken.None);
|
commitMessage: "", ct: CancellationToken.None);
|
||||||
|
|
||||||
Assert.Equal("merged", result.Status);
|
Assert.Equal("merged", result.Status);
|
||||||
Assert.Contains(proxy.Calls, c => c.Method == "MergeProgress"
|
Assert.Contains(proxy.Calls, c => c.Method == "OperationProgress"
|
||||||
&& (string?)c.Args[0] == task.Id && (string?)c.Args[1] == TaskMergeService.PhaseMerging);
|
&& (string?)c.Args[0] == task.Id && (string?)c.Args[1] == TaskMergeService.PhaseMerging);
|
||||||
Assert.Contains(proxy.Calls, c => c.Method == "MergeProgress"
|
Assert.Contains(proxy.Calls, c => c.Method == "OperationProgress"
|
||||||
&& (string?)c.Args[0] == task.Id && (string?)c.Args[1] == TaskMergeService.PhaseVerifying);
|
&& (string?)c.Args[0] == task.Id && (string?)c.Args[1] == TaskMergeService.PhaseVerifying);
|
||||||
Assert.Contains(proxy.Calls, c => c.Method == "WorkerLog"
|
Assert.Contains(proxy.Calls, c => c.Method == "WorkerLog"
|
||||||
&& c.Args[0] is string s && s.Contains("Verify command running"));
|
&& c.Args[0] is string s && s.Contains("Verify command running"));
|
||||||
|
|||||||
@@ -113,6 +113,7 @@ sealed class FakeWorkerClient : IWorkerClient
|
|||||||
public event Action<bool>? PrepFinishedEvent;
|
public event Action<bool>? PrepFinishedEvent;
|
||||||
public event Action<string>? RefineStartedEvent;
|
public event Action<string>? RefineStartedEvent;
|
||||||
public event Action<string, bool, string?>? RefineFinishedEvent;
|
public event Action<string, bool, string?>? RefineFinishedEvent;
|
||||||
|
public event Action<string, string, int, int>? OperationProgressEvent;
|
||||||
public event Action<string, string, int>? MergeProgressEvent;
|
public event Action<string, string, int>? MergeProgressEvent;
|
||||||
public event Action<string, string>? PlanningMergeStartedEvent;
|
public event Action<string, string>? PlanningMergeStartedEvent;
|
||||||
public event Action<string, string>? PlanningSubtaskMergedEvent;
|
public event Action<string, string>? PlanningSubtaskMergedEvent;
|
||||||
|
|||||||
Reference in New Issue
Block a user