CanRemoveFromQueue schloss IsWaiting (queued+blocked) faelschlich aus, sodass ein Kettenglied in der Mitte nicht entfernbar war. Zusaetzlich unblockt TaskStateService.DequeueToIdleAsync jetzt den Nachfolger eines dequeueten Kettenglieds, sonst bliebe er dauerhaft auf einen nicht mehr laufenden Vorgaenger blockiert.
743 lines
33 KiB
C#
743 lines
33 KiB
C#
using ClaudeDo.Data;
|
|
using ClaudeDo.Data.Models;
|
|
using ClaudeDo.Data.Repositories;
|
|
using ClaudeDo.Worker.Git;
|
|
using ClaudeDo.Worker.Hub;
|
|
using ClaudeDo.Worker.Planning;
|
|
using ClaudeDo.Worker.Queue;
|
|
using ClaudeDo.Worker.Tickets;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
|
|
|
|
namespace ClaudeDo.Worker.State;
|
|
|
|
public sealed class TaskStateService : ITaskStateService
|
|
{
|
|
private readonly IDbContextFactory<ClaudeDoDbContext> _dbFactory;
|
|
private readonly HubBroadcaster _broadcaster;
|
|
private readonly IQueueWaker _waker;
|
|
private readonly PlanningChainCoordinator _chain;
|
|
private readonly RunCancellationRegistry _runCancels;
|
|
private readonly Func<IActiveMergeState> _mergeState;
|
|
private readonly BaseDirtyChecker _baseDirtyChecker;
|
|
private readonly ILogger<TaskStateService> _logger;
|
|
private readonly TicketStatusSync _ticketSync;
|
|
|
|
public TaskStateService(
|
|
IDbContextFactory<ClaudeDoDbContext> dbFactory,
|
|
HubBroadcaster broadcaster,
|
|
IQueueWaker waker,
|
|
PlanningChainCoordinator chain,
|
|
RunCancellationRegistry runCancels,
|
|
Func<IActiveMergeState> mergeState,
|
|
BaseDirtyChecker baseDirtyChecker,
|
|
ILogger<TaskStateService> logger,
|
|
TicketStatusSync ticketSync)
|
|
{
|
|
_dbFactory = dbFactory;
|
|
_broadcaster = broadcaster;
|
|
_waker = waker;
|
|
_chain = chain;
|
|
_runCancels = runCancels;
|
|
_mergeState = mergeState;
|
|
_baseDirtyChecker = baseDirtyChecker;
|
|
_logger = logger;
|
|
_ticketSync = ticketSync;
|
|
}
|
|
|
|
public async Task<TransitionResult> EnqueueAsync(string taskId, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
|
|
if (await IsDraftChildAsync(ctx, taskId, ct))
|
|
return new TransitionResult(false, "Draft subtask: finalize the plan before queuing it.");
|
|
|
|
if (await IsManualAsync(ctx, taskId, ct))
|
|
return new TransitionResult(false, "Manual task: mark it as a Claude task before queuing it.");
|
|
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId && t.Status != TaskStatus.Running)
|
|
.ExecuteUpdateAsync(s => s.SetProperty(t => t.Status, TaskStatus.Queued), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task not found or already running.");
|
|
|
|
_waker.Wake();
|
|
await NotifyAsync(taskId);
|
|
|
|
// Heads-up only: a worktree forks from the commit tip (WorktreeManager.ResolveBaseCommitAsync),
|
|
// not the working tree, so uncommitted changes sitting in the list's repo right now are
|
|
// invisible to this (and every other) queued task. Never blocks the transition above.
|
|
var workingDir = await ctx.Tasks.AsNoTracking()
|
|
.Where(t => t.Id == taskId)
|
|
.Select(t => t.List.WorkingDir)
|
|
.FirstOrDefaultAsync(ct);
|
|
var baseDirty = await _baseDirtyChecker.CheckAsync(workingDir, ct);
|
|
|
|
return new TransitionResult(true, null, baseDirty);
|
|
}
|
|
|
|
public async Task<TransitionResult> StartRunningAsync(string taskId, DateTime startedAt, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
|
|
if (await IsDraftChildAsync(ctx, taskId, ct))
|
|
return new TransitionResult(false, "Draft subtask: finalize the plan before running it.");
|
|
|
|
if (await IsManualAsync(ctx, taskId, ct))
|
|
return new TransitionResult(false, "Manual task: mark it as a Claude task before running it.");
|
|
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId && t.Status != TaskStatus.Running)
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, TaskStatus.Running)
|
|
.SetProperty(t => t.StartedAt, startedAt), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task already running or not found.");
|
|
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> CompleteAsync(string taskId, DateTime finishedAt, string? result, CancellationToken ct)
|
|
{
|
|
await using (var ctx = await _dbFactory.CreateDbContextAsync(ct))
|
|
{
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId && t.Status == TaskStatus.Running)
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, TaskStatus.Done)
|
|
.SetProperty(t => t.FinishedAt, finishedAt)
|
|
.SetProperty(t => t.Result, result), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task not running; cannot complete.");
|
|
}
|
|
|
|
await OnChildTerminalAsync(taskId, TaskStatus.Done);
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> SubmitForReviewAsync(string taskId, DateTime finishedAt, string? result, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId && t.Status == TaskStatus.Running)
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, TaskStatus.WaitingForReview)
|
|
.SetProperty(t => t.FinishedAt, finishedAt)
|
|
.SetProperty(t => t.Result, result), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task not running; cannot submit for review.");
|
|
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
// Submit an interactively-worked task (a ConPTY session left its worktree with commits/changes)
|
|
// for review. Unlike SubmitForReviewAsync — which only fires from the headless Running state —
|
|
// this transitions from Idle or Failed, the states an interactive task sits in after the user
|
|
// finishes the session by hand. The caller commits the worktree first so there is a diff to merge.
|
|
public async Task<TransitionResult> SubmitInteractiveForReviewAsync(string taskId, DateTime finishedAt, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId && (t.Status == TaskStatus.Idle || t.Status == TaskStatus.Failed))
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, TaskStatus.WaitingForReview)
|
|
.SetProperty(t => t.FinishedAt, finishedAt), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task is not Idle or Failed; cannot submit for review.");
|
|
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> SubmitForChildrenAsync(string taskId, DateTime finishedAt, string? result, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId && t.Status == TaskStatus.Running)
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, TaskStatus.WaitingForChildren)
|
|
.SetProperty(t => t.FinishedAt, finishedAt)
|
|
.SetProperty(t => t.Result, result), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task not running; cannot submit for children.");
|
|
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> ApproveReviewAsync(string taskId, CancellationToken ct)
|
|
{
|
|
var now = DateTime.UtcNow;
|
|
await using (var ctx = await _dbFactory.CreateDbContextAsync(ct))
|
|
{
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId && t.Status == TaskStatus.WaitingForReview)
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, TaskStatus.Done)
|
|
.SetProperty(t => t.FinishedAt, now), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task is not waiting for review; cannot approve.");
|
|
}
|
|
|
|
await OnChildTerminalAsync(taskId, TaskStatus.Done);
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> RejectToQueueAsync(string taskId, string feedback, CancellationToken ct)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(feedback))
|
|
return new TransitionResult(false, "Feedback is required to reject for re-run.");
|
|
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId && t.Status == TaskStatus.WaitingForReview)
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, TaskStatus.Queued)
|
|
.SetProperty(t => t.ReviewFeedback, feedback)
|
|
.SetProperty(t => t.StartedAt, (DateTime?)null)
|
|
.SetProperty(t => t.FinishedAt, (DateTime?)null), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task is not waiting for review; cannot reject.");
|
|
|
|
_waker.Wake();
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> RejectToIdleAsync(string taskId, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId && t.Status == TaskStatus.WaitingForReview)
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, TaskStatus.Idle)
|
|
.SetProperty(t => t.ReviewFeedback, (string?)null), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task is not waiting for review; cannot park.");
|
|
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> ClearReviewFeedbackAsync(string taskId, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId)
|
|
.ExecuteUpdateAsync(s => s.SetProperty(t => t.ReviewFeedback, (string?)null), ct);
|
|
|
|
return affected == 0
|
|
? new TransitionResult(false, "Task not found.")
|
|
: new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> FailAsync(
|
|
string taskId, DateTime finishedAt, string? error, CancellationToken ct,
|
|
string failureReason = "error", int? turnsUsed = null, int? maxTurns = null)
|
|
{
|
|
await using (var ctx = await _dbFactory.CreateDbContextAsync(ct))
|
|
{
|
|
// Queued is intentional: OverrideSlotService dispatches RunAsync before calling
|
|
// StartRunningAsync, so a preflight failure (list not found, worktree setup) can
|
|
// reach MarkFailed while the task is still Queued in the DB.
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId &&
|
|
(t.Status == TaskStatus.Running || t.Status == TaskStatus.Queued))
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, TaskStatus.Failed)
|
|
.SetProperty(t => t.FinishedAt, finishedAt)
|
|
.SetProperty(t => t.Result, error)
|
|
.SetProperty(t => t.FailureReason, failureReason)
|
|
.SetProperty(t => t.FailureTurnsUsed, turnsUsed)
|
|
.SetProperty(t => t.FailureMaxTurns, maxTurns), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task not in a failable state (must be Running or Queued).");
|
|
}
|
|
|
|
await OnChildTerminalAsync(taskId, TaskStatus.Failed);
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
// allowFromIdle: only the external update_task_status(Cancelled) path sets this — it lets
|
|
// an Idle task be retired without deleting it. Every other caller (hub CancelReview,
|
|
// PlanningChainCoordinator's chain-walk, batch cancel) relies on Idle staying a no-op here;
|
|
// PlanningChainCoordinator specifically uses "parked back to Idle" as a deliberate opt-out
|
|
// signal, so do not flip this default.
|
|
public async Task<TransitionResult> CancelAsync(
|
|
string taskId, DateTime finishedAt, CancellationToken ct, bool allowFromIdle = false)
|
|
{
|
|
// A unit merge drains its children's worktrees onto the target branch while the parent
|
|
// sits in WaitingForReview for the whole (potentially minutes-long) drain. Cancelling out
|
|
// from under it would flip the parent to Cancelled while the orchestrator keeps merging —
|
|
// FinalizeParentDoneAsync then finds the parent no longer WaitingForReview and gives up,
|
|
// leaving the merged children's diffs stranded with no way to roll them back.
|
|
if (await _mergeState().HasActiveMergeAsync(taskId, ct))
|
|
return new TransitionResult(false, "A merge is in progress for this task; wait for it to finish before cancelling.");
|
|
|
|
List<string> cancelledChildIds;
|
|
await using (var ctx = await _dbFactory.CreateDbContextAsync(ct))
|
|
{
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId &&
|
|
(t.Status == TaskStatus.Running || t.Status == TaskStatus.Queued
|
|
|| t.Status == TaskStatus.WaitingForReview
|
|
|| t.Status == TaskStatus.WaitingForChildren
|
|
|| (allowFromIdle && t.Status == TaskStatus.Idle)))
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, TaskStatus.Cancelled)
|
|
.SetProperty(t => t.FinishedAt, finishedAt), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task not in cancellable state.");
|
|
|
|
// Cascade to this task's own non-terminal children (one level — children are
|
|
// leaves in the current model) so a Queued/Running child doesn't keep going
|
|
// against a parent that's already gone, orphaning its worktree commit.
|
|
cancelledChildIds = await ctx.Tasks
|
|
.Where(t => t.ParentTaskId == taskId
|
|
&& t.Status != TaskStatus.Done
|
|
&& t.Status != TaskStatus.Failed
|
|
&& t.Status != TaskStatus.Cancelled)
|
|
.Select(t => t.Id)
|
|
.ToListAsync(ct);
|
|
|
|
if (cancelledChildIds.Count > 0)
|
|
await ctx.Tasks
|
|
.Where(t => cancelledChildIds.Contains(t.Id))
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, TaskStatus.Cancelled)
|
|
.SetProperty(t => t.FinishedAt, finishedAt)
|
|
.SetProperty(t => t.BlockedByTaskId, (string?)null), ct);
|
|
}
|
|
|
|
// Also stop the actual Claude processes: the DB flip above doesn't reach a run
|
|
// already executing in a queue/override slot — without this it would keep going
|
|
// and only fail its final (now-invalid) transition.
|
|
_runCancels.TryCancel(taskId);
|
|
foreach (var childId in cancelledChildIds)
|
|
_runCancels.TryCancel(childId);
|
|
|
|
await OnChildTerminalAsync(taskId, TaskStatus.Cancelled);
|
|
await NotifyAsync(taskId);
|
|
foreach (var childId in cancelledChildIds)
|
|
await NotifyAsync(childId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> ResetToIdleAsync(string taskId, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId && t.Status != TaskStatus.Running)
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, TaskStatus.Idle)
|
|
.SetProperty(t => t.StartedAt, (DateTime?)null)
|
|
.SetProperty(t => t.FinishedAt, (DateTime?)null)
|
|
.SetProperty(t => t.Result, (string?)null), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task is running; cannot reset.");
|
|
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
// Unconditional status write — bypasses transition rules. Used by the UI's
|
|
// "set status freely" affordance; intentionally no guards (caller may strand
|
|
// the runner if used while a task is executing). It also bypasses chain/parent
|
|
// advancement — forcing a chain child or a WaitingForChildren parent's child to a
|
|
// terminal status here does not unblock its successor or re-check the parent.
|
|
public async Task<TransitionResult> ForceSetStatusAsync(string taskId, TaskStatus status, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
var affected = status == TaskStatus.Done
|
|
? await ctx.Tasks
|
|
.Where(t => t.Id == taskId)
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, status)
|
|
.SetProperty(t => t.FinishedAt, DateTime.UtcNow), ct)
|
|
: await ctx.Tasks
|
|
.Where(t => t.Id == taskId)
|
|
.ExecuteUpdateAsync(s => s.SetProperty(t => t.Status, status), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task not found.");
|
|
|
|
if (status == TaskStatus.Queued) _waker.Wake();
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> MarkDoneAsync(string taskId, DateTime finishedAt, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
// The done toggle is a manual "tick it off" affordance and works from every status
|
|
// except Running (a live run keeps writing the row, so the picker/runner would race).
|
|
// WaitingForReview is deliberately included: ticking a finished task done skips the
|
|
// merge on purpose -- for a worktree-less list-handler task there is nothing to merge,
|
|
// and for a worktree task the branch just stays Active for the worktrees overview.
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId && t.Status != TaskStatus.Running)
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, TaskStatus.Done)
|
|
.SetProperty(t => t.FinishedAt, finishedAt), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task is running or no longer exists; cannot mark done.");
|
|
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> UnmarkDoneAsync(string taskId, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId && t.Status == TaskStatus.Done)
|
|
.ExecuteUpdateAsync(s => s.SetProperty(t => t.Status, TaskStatus.Idle), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task is not Done; cannot unmark.");
|
|
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> DequeueToIdleAsync(string taskId, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId && t.Status == TaskStatus.Queued)
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, TaskStatus.Idle)
|
|
.SetProperty(t => t.BlockedByTaskId, (string?)null), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task is not queued; cannot remove from queue.");
|
|
|
|
// A dequeued task may have been a planning-chain predecessor. Without unblocking its
|
|
// successor here, the successor is left permanently wedged on a BlockedByTaskId link to
|
|
// a task that has left the queue and will never reach the terminal state that would
|
|
// otherwise resolve it (PlanningChainCoordinator.OnChildFinishedAsync).
|
|
var successorId = await ctx.Tasks.AsNoTracking()
|
|
.Where(t => t.BlockedByTaskId == taskId)
|
|
.Select(t => t.Id)
|
|
.FirstOrDefaultAsync(ct);
|
|
if (successorId is not null)
|
|
await UnblockAsync(successorId, ct);
|
|
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> StartPlanningAsync(string parentId, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == parentId
|
|
&& t.Status == TaskStatus.Idle
|
|
&& t.PlanningPhase == PlanningPhase.None)
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.PlanningPhase, PlanningPhase.Active), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task not in plannable state.");
|
|
|
|
await NotifyAsync(parentId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> FinalizePlanningAsync(string parentId, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
var hasChildren = await ctx.Tasks.AnyAsync(t => t.ParentTaskId == parentId, ct);
|
|
var newStatus = hasChildren ? TaskStatus.WaitingForChildren : TaskStatus.WaitingForReview;
|
|
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == parentId && t.PlanningPhase == PlanningPhase.Active)
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.PlanningPhase, PlanningPhase.Finalized)
|
|
.SetProperty(t => t.PlanningFinalizedAt, DateTime.UtcNow)
|
|
.SetProperty(t => t.PlanningSessionToken, (string?)null)
|
|
.SetProperty(t => t.Status, newStatus), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "No active planning session.");
|
|
|
|
await NotifyAsync(parentId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> BlockOnAsync(string taskId, string predecessorTaskId, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId)
|
|
.ExecuteUpdateAsync(s => s.SetProperty(t => t.BlockedByTaskId, predecessorTaskId), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task not found.");
|
|
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> UnblockAsync(string taskId, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
var affected = await ctx.Tasks
|
|
.Where(t => t.Id == taskId)
|
|
.ExecuteUpdateAsync(s => s.SetProperty(t => t.BlockedByTaskId, (string?)null), ct);
|
|
|
|
if (affected == 0)
|
|
return new TransitionResult(false, "Task not found.");
|
|
|
|
_waker.Wake();
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
public async Task<TransitionResult> SetDependsOnAsync(
|
|
string taskId, IReadOnlyList<string> dependsOnTaskIds, CancellationToken ct)
|
|
{
|
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
|
|
|
if (!await ctx.Tasks.AsNoTracking().AnyAsync(t => t.Id == taskId, ct))
|
|
return new TransitionResult(false, "Task not found.");
|
|
|
|
var wanted = dependsOnTaskIds.Distinct(StringComparer.Ordinal).ToList();
|
|
|
|
if (wanted.Count > 0)
|
|
{
|
|
if (wanted.Contains(taskId, StringComparer.Ordinal))
|
|
return new TransitionResult(false, "A task cannot depend on itself.");
|
|
|
|
var known = await ctx.Tasks.AsNoTracking()
|
|
.Where(t => wanted.Contains(t.Id))
|
|
.Select(t => t.Id)
|
|
.ToListAsync(ct);
|
|
var missing = wanted.Except(known, StringComparer.Ordinal).FirstOrDefault();
|
|
if (missing is not null)
|
|
return new TransitionResult(false, $"Dependency task {missing} not found.");
|
|
|
|
if (await WouldCycleAsync(ctx, taskId, wanted, ct))
|
|
return new TransitionResult(false, "Setting this dependency would create a cycle.");
|
|
}
|
|
|
|
var existing = await ctx.TaskDependencies.AsNoTracking()
|
|
.Where(d => d.TaskId == taskId)
|
|
.Select(d => d.DependsOnTaskId)
|
|
.ToListAsync(ct);
|
|
|
|
var removed = existing.Except(wanted, StringComparer.Ordinal).ToList();
|
|
var added = wanted.Except(existing, StringComparer.Ordinal).ToList();
|
|
if (removed.Count == 0 && added.Count == 0)
|
|
return new TransitionResult(true, null);
|
|
|
|
if (removed.Count > 0)
|
|
await ctx.TaskDependencies
|
|
.Where(d => d.TaskId == taskId && removed.Contains(d.DependsOnTaskId))
|
|
.ExecuteDeleteAsync(ct);
|
|
|
|
if (added.Count > 0)
|
|
{
|
|
ctx.TaskDependencies.AddRange(added.Select(id =>
|
|
new TaskDependencyEntity { TaskId = taskId, DependsOnTaskId = id }));
|
|
await ctx.SaveChangesAsync(ct);
|
|
}
|
|
|
|
// Dropping an edge may free up a Queued task the picker was skipping.
|
|
if (removed.Count > 0) _waker.Wake();
|
|
await NotifyAsync(taskId);
|
|
return new TransitionResult(true, null);
|
|
}
|
|
|
|
// True when any proposed predecessor can already reach taskId through the dependency graph --
|
|
// linking there would starve both ends forever, since the picker never claims either.
|
|
// ponytail: loads the whole edge set (only *declared* dependencies land there, so it stays
|
|
// small). Swap in a recursive CTE if task_dependencies ever grows past a few thousand rows.
|
|
private static async Task<bool> WouldCycleAsync(
|
|
ClaudeDoDbContext ctx, string taskId, IReadOnlyList<string> proposed, CancellationToken ct)
|
|
{
|
|
var edges = await ctx.TaskDependencies.AsNoTracking()
|
|
.Select(d => new { d.TaskId, d.DependsOnTaskId })
|
|
.ToListAsync(ct);
|
|
var successors = edges
|
|
.GroupBy(e => e.TaskId, StringComparer.Ordinal)
|
|
.ToDictionary(g => g.Key, g => g.Select(e => e.DependsOnTaskId).ToList(), StringComparer.Ordinal);
|
|
|
|
// `visited` also stops the walk looping forever on pre-existing bad data.
|
|
var visited = new HashSet<string>(StringComparer.Ordinal);
|
|
var frontier = new Queue<string>(proposed);
|
|
while (frontier.Count > 0)
|
|
{
|
|
var current = frontier.Dequeue();
|
|
if (current == taskId) return true;
|
|
if (!visited.Add(current)) continue;
|
|
if (successors.TryGetValue(current, out var next))
|
|
foreach (var n in next) frontier.Enqueue(n);
|
|
}
|
|
return false;
|
|
}
|
|
|
|
public async Task<int> RecoverStaleRunningAsync(string reason, CancellationToken ct)
|
|
{
|
|
var resultText = "[stale] " + reason;
|
|
var now = DateTime.UtcNow;
|
|
List<string> recoveredIds;
|
|
int affected;
|
|
await using (var ctx = await _dbFactory.CreateDbContextAsync(ct))
|
|
{
|
|
recoveredIds = await ctx.Tasks
|
|
.Where(t => t.Status == TaskStatus.Running)
|
|
.Select(t => t.Id)
|
|
.ToListAsync(ct);
|
|
|
|
affected = await ctx.Tasks
|
|
.Where(t => t.Status == TaskStatus.Running)
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, TaskStatus.Failed)
|
|
.SetProperty(t => t.FinishedAt, now)
|
|
.SetProperty(t => t.Result, resultText), ct);
|
|
}
|
|
|
|
// A recovered task may have been a planning/improvement chain child or the last
|
|
// non-terminal child of a WaitingForChildren parent. The bulk flip above skips the
|
|
// usual terminal-transition side effects, so without this a crash mid-run would
|
|
// leave the chain successor blocked forever and the parent wedged in
|
|
// WaitingForChildren with nothing left to re-check it.
|
|
foreach (var taskId in recoveredIds)
|
|
await OnChildTerminalAsync(taskId, TaskStatus.Failed);
|
|
|
|
return affected;
|
|
}
|
|
|
|
// A subtask is "draft" only while its planning parent has an open (Active) session.
|
|
// Improvement children whose parent has PlanningPhase.None are not drafts and may be
|
|
// queued freely. Standalone tasks (no parent) are never draft.
|
|
// Server-side backstop for the manual flag: the UI hides the hand-off affordances, but the
|
|
// MCP surface and hub can still be driven directly.
|
|
private static Task<bool> IsManualAsync(ClaudeDoDbContext ctx, string taskId, CancellationToken ct)
|
|
=> ctx.Tasks.AsNoTracking().AnyAsync(t => t.Id == taskId && t.IsManual, ct);
|
|
|
|
private static async Task<bool> IsDraftChildAsync(ClaudeDoDbContext ctx, string taskId, CancellationToken ct)
|
|
{
|
|
var parentId = await ctx.Tasks.AsNoTracking()
|
|
.Where(t => t.Id == taskId)
|
|
.Select(t => t.ParentTaskId)
|
|
.FirstOrDefaultAsync(ct);
|
|
if (parentId is null) return false;
|
|
|
|
return await ctx.Tasks.AsNoTracking()
|
|
.AnyAsync(p => p.Id == parentId && p.PlanningPhase == PlanningPhase.Active, ct);
|
|
}
|
|
|
|
private async Task OnChildTerminalAsync(string taskId, TaskStatus finalStatus)
|
|
{
|
|
// Terminal child writes are best-effort and use CancellationToken.None so the
|
|
// task lifecycle is never left partially completed because a caller cancelled.
|
|
string? parentId;
|
|
await using (var ctx = await _dbFactory.CreateDbContextAsync(CancellationToken.None))
|
|
{
|
|
parentId = await ctx.Tasks
|
|
.AsNoTracking()
|
|
.Where(t => t.Id == taskId)
|
|
.Select(t => t.ParentTaskId)
|
|
.FirstOrDefaultAsync(CancellationToken.None);
|
|
}
|
|
if (parentId is null) return;
|
|
|
|
try
|
|
{
|
|
await _chain.OnChildFinishedAsync(taskId, finalStatus, CancellationToken.None);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "PlanningChain advance failed for {TaskId}", taskId);
|
|
}
|
|
|
|
await TryAdvanceParentAsync(parentId);
|
|
}
|
|
|
|
// Any parent (planning or improvement) sitting in WaitingForChildren surfaces for review
|
|
// once every child is terminal (Done/Failed/Cancelled). A failed or cancelled child does
|
|
// not wedge the parent — it is flagged on the result. Also called directly after a child
|
|
// is deleted, since no terminal transition fires in that case.
|
|
public async Task TryAdvanceParentAsync(string parentId)
|
|
{
|
|
try
|
|
{
|
|
await AdvanceParentIfAllChildrenTerminalAsync(parentId);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "TryAdvanceParent failed for {ParentId}", parentId);
|
|
}
|
|
}
|
|
|
|
private async Task AdvanceParentIfAllChildrenTerminalAsync(string parentId)
|
|
{
|
|
string? parentResult;
|
|
List<TaskStatus> childStatuses;
|
|
await using (var ctx = await _dbFactory.CreateDbContextAsync(CancellationToken.None))
|
|
{
|
|
var parent = await ctx.Tasks.AsNoTracking()
|
|
.FirstOrDefaultAsync(t => t.Id == parentId, CancellationToken.None);
|
|
if (parent is null || parent.Status != TaskStatus.WaitingForChildren) return;
|
|
parentResult = parent.Result;
|
|
childStatuses = await ctx.Tasks
|
|
.Where(t => t.ParentTaskId == parentId)
|
|
.Select(t => t.Status)
|
|
.ToListAsync(CancellationToken.None);
|
|
}
|
|
// No early-out on an empty list: zero children left (e.g. the last one was just
|
|
// deleted) counts as "all terminal" — .All() on an empty sequence is vacuously true.
|
|
bool allTerminal = childStatuses.All(s =>
|
|
s == TaskStatus.Done || s == TaskStatus.Failed || s == TaskStatus.Cancelled);
|
|
if (!allTerminal) return;
|
|
|
|
int failed = childStatuses.Count(s => s == TaskStatus.Failed);
|
|
int cancelled = childStatuses.Count(s => s == TaskStatus.Cancelled);
|
|
var newResult = parentResult;
|
|
if (failed + cancelled > 0)
|
|
{
|
|
var note = $"⚠ Children: {failed} failed, {cancelled} cancelled.";
|
|
newResult = string.IsNullOrWhiteSpace(parentResult) ? note : $"{parentResult}\n\n{note}";
|
|
}
|
|
|
|
await using var writeCtx = await _dbFactory.CreateDbContextAsync(CancellationToken.None);
|
|
await writeCtx.Tasks
|
|
.Where(t => t.Id == parentId && t.Status == TaskStatus.WaitingForChildren)
|
|
.ExecuteUpdateAsync(s => s
|
|
.SetProperty(t => t.Status, TaskStatus.WaitingForReview)
|
|
.SetProperty(t => t.Result, newResult), CancellationToken.None);
|
|
await NotifyAsync(parentId);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Der einzige Broadcast-Pfad dieser Klasse. Zusätzlich zum UI-Update meldet er den neuen
|
|
/// Status ans verknüpfte Ticket. Nie direkt _broadcaster.TaskUpdated aufrufen — sonst
|
|
/// fällt genau dieser Übergang aus dem Ticket-Sync heraus.
|
|
/// </summary>
|
|
private async Task NotifyAsync(string taskId)
|
|
{
|
|
await _broadcaster.TaskUpdated(taskId);
|
|
await _ticketSync.SyncAsync(taskId, CancellationToken.None);
|
|
}
|
|
}
|