feat(claude-do): depends_on als Menge statt Einzelfeld
Ein Task kann jetzt auf beliebig viele Vorgaenger warten. Die Einzelspalte tasks.depends_on_task_id wird durch die Beziehungstabelle task_dependencies ersetzt (Composite-PK, beide FKs ON DELETE CASCADE -- das reproduziert das alte ON DELETE SET NULL: geloeschter Vorgaenger = Kante weg = Task wieder pickbar). BlockedByTaskId bleibt unberuehrt. - Migration AddTaskDependencies kopiert jede bestehende depends_on_task_id verlustfrei in die Tabelle, bevor die Spalte fallengelassen wird - QueuePicker gatet per doppelt negiertem NOT EXISTS: kein Task mit 0 Kanten wird blockiert, alle Kanten muessen done sein, eine Kante auf einen verschwundenen Task blockiert weiter - SetDependsOnAsync ersetzt die ganze Menge; Zykluspruefung ist jetzt eine BFS ueber den Graphen, faengt also auch A->B->C->A - blockedReason nennt alle offenen Vorgaenger (External/TaskBlocking, von ExternalMcpService und TaskWaitMcpTools gemeinsam genutzt) - add_task/update_task nehmen dependsOnTaskIds; die alte Skalarform bleibt als Ein-Element-Menge inkl. ""-Clear-Sentinel gueltig - UI: Fan-in-Zeile bleibt flach und listet alle Vorgaenger im After-Label, die lineare Kette rendert unveraendert
This commit is contained in:
@@ -50,8 +50,8 @@ subfolder within their area; the namespace stays the area namespace.
|
||||
## Architecture
|
||||
|
||||
- **Program.cs** — loads config, inits schema, registers DI, configures SignalR on `/hub`, binds to `127.0.0.1:47821`
|
||||
- **TaskStateService** — the **only** component that writes `Status`, `PlanningPhase`, `BlockedByTaskId`, `DependsOnTaskId`. All transitions return a `TransitionResult` (no exceptions on invalid moves). Wakes the queue and advances the planning chain on child terminal transitions. Every transition broadcasts via a private `NotifyAsync(taskId)` — never call `_broadcaster.TaskUpdated` directly from this class, `NotifyAsync` also drives `TicketStatusSync.SyncAsync` (see Tickets below) and a direct call skips that transition's ticket write-back. `SetDependsOnAsync` rejects a self-reference, an unknown dependency id, or a link that would create a cycle (walks the proposed predecessor's own `DependsOnTaskId` chain).
|
||||
- **IQueueWaker / IQueuePicker / QueueService** — waker is a singleton `SemaphoreSlim`; picker performs the atomic `Queued → Running` claim filtered by `BlockedByTaskId IS NULL`, `is_manual = 0`, schedule, and (`DependsOnTaskId IS NULL` OR the dependency's `Status = 'done'`); QueueService is a thin `BackgroundService` looping on the waker, dispatching via `TaskRunner`. Per tick it also applies the usage throttle and gate → [usage-monitoring](../../docs/explore-notes/usage-monitoring.md).
|
||||
- **TaskStateService** — the **only** component that writes `Status`, `PlanningPhase`, `BlockedByTaskId`, and the `task_dependencies` edges. All transitions return a `TransitionResult` (no exceptions on invalid moves). Wakes the queue and advances the planning chain on child terminal transitions. Every transition broadcasts via a private `NotifyAsync(taskId)` — never call `_broadcaster.TaskUpdated` directly from this class, `NotifyAsync` also drives `TicketStatusSync.SyncAsync` (see Tickets below) and a direct call skips that transition's ticket write-back. `SetDependsOnAsync` replaces the task's whole dependency **set** (empty list clears it) and rejects a self-reference, an unknown dependency id, or any link that would create a cycle — a BFS over the whole `task_dependencies` edge set from every proposed predecessor, so a multi-edge loop (A→B→C→A) is caught too, and one bad edge leaves the entire set unwritten.
|
||||
- **IQueueWaker / IQueuePicker / QueueService** — waker is a singleton `SemaphoreSlim`; picker performs the atomic `Queued → Running` claim filtered by `BlockedByTaskId IS NULL`, `is_manual = 0`, schedule, and "no `task_dependencies` edge whose predecessor isn't `done`" (doubly negated on purpose: zero edges stays eligible, all edges must be done, and an edge pointing at a vanished task keeps blocking); QueueService is a thin `BackgroundService` looping on the waker, dispatching via `TaskRunner`. Per tick it also applies the usage throttle and gate → [usage-monitoring](../../docs/explore-notes/usage-monitoring.md).
|
||||
- **RunCancellationRegistry** — taskId → running-run CTS. Lets `TaskStateService.CancelAsync` kill a cancelled task's process without a DI cycle.
|
||||
- **OverrideSlotService** — owns `RunNow` / `ContinueTask`; goes through `TaskStateService.StartRunningAsync` (caller-driven, serialized by slot lock).
|
||||
- **StaleTaskRecovery** — startup-only; calls `TaskStateService.RecoverStaleRunningAsync` to flip orphaned `Running` rows to `Failed`.
|
||||
@@ -67,7 +67,7 @@ not conflated.
|
||||
| `Status` | `Idle`, `Queued`, `Running`, `WaitingForChildren`, `WaitingForReview`, `Done`, `Failed`, `Cancelled` | Lifecycle only. `WaitingForChildren` = parent's own work done, waiting on children. |
|
||||
| `PlanningPhase` | `None`, `Active`, `Finalized` | Parent-only marker. `Active` ≈ legacy `Planning`; `Finalized` ≈ legacy `Planned`. |
|
||||
| `BlockedByTaskId` | nullable FK | Replaces legacy `Waiting`. A queued row with a non-null value is skipped by the picker. Internal to `PlanningChainCoordinator` — resolves (or cascades) on ANY terminal state of the predecessor. Because it self-resolves, `wait_for_task_change` deliberately keeps **waiting** through it instead of reporting `"Blocked"`. |
|
||||
| `DependsOnTaskId` | nullable FK | User/MCP-declared predecessor (`add_task`/`update_task`), separate from `BlockedByTaskId` because the semantics differ: the picker only skips a queued row while the dependency's `Status` isn't `Done` -- a Failed/Cancelled dependency does **not** cascade or auto-resolve, the dependent just stays blocked (see `QueuePicker`, `TaskStateService.SetDependsOnAsync`). `get_task`/`list_tasks`/`batch_get_tasks` surface this as `blocked`/`blockedReason`; `wait_for_task_change` reports `"Blocked"` instead of silently running out its timeout. |
|
||||
| `Dependencies` | 0..n rows in `task_dependencies` | User/MCP-declared predecessors (`add_task`/`update_task`), separate from `BlockedByTaskId` because the semantics differ: the picker skips a queued row until **every** dependency reached `Done` -- a Failed/Cancelled dependency does **not** cascade or auto-resolve, the dependent just stays blocked (see `QueuePicker`, `TaskStateService.SetDependsOnAsync`). `get_task`/`list_tasks`/`batch_get_tasks` surface this as `dependsOnTaskIds` + `blocked`/`blockedReason` (the reason names **all** open predecessors, via `External/TaskBlocking`); `wait_for_task_change` reports `"Blocked"` instead of silently running out its timeout. |
|
||||
| `IsManual` | bool | Reminder only the user can do. `EnqueueAsync`/`StartRunningAsync` refuse it, the picker skips it, `GetDailyPrepCandidates` never offers it. An interactive ConPTY session is still allowed. |
|
||||
| `ReviewFeedback` | nullable string | Reviewer's rejection comment; consumed and cleared by `QueueService` on the next re-run. |
|
||||
|
||||
|
||||
+67
-39
@@ -1,4 +1,4 @@
|
||||
using System.ComponentModel;
|
||||
using System.ComponentModel;
|
||||
using System.Diagnostics;
|
||||
using System.Globalization;
|
||||
using System.Text;
|
||||
@@ -55,11 +55,11 @@ public sealed record TaskDto(
|
||||
string? FailureReason = null,
|
||||
int? FailureTurnsUsed = null,
|
||||
int? FailureMaxTurns = null,
|
||||
// A user/MCP-declared predecessor (set via add_task/update_task), distinct from the
|
||||
// planning chain's own internal BlockedByTaskId link.
|
||||
string? DependsOnTaskId = null,
|
||||
// User/MCP-declared predecessors (set via add_task/update_task), distinct from the planning
|
||||
// chain's own internal BlockedByTaskId link. Empty unless the task declares dependencies.
|
||||
IReadOnlyList<string>? DependsOnTaskIds = null,
|
||||
// True only while Status is Queued and the picker will not claim this task yet -- either a
|
||||
// planning-chain predecessor or DependsOnTaskId hasn't reached Done. See BlockedReason.
|
||||
// planning-chain predecessor or some dependsOnTaskIds entry hasn't reached Done. See BlockedReason.
|
||||
bool Blocked = false,
|
||||
string? BlockedReason = null,
|
||||
bool IsManual = false,
|
||||
@@ -82,7 +82,7 @@ public sealed record TaskRefDto(
|
||||
string? FailureReason = null,
|
||||
int? FailureTurnsUsed = null,
|
||||
int? FailureMaxTurns = null,
|
||||
string? DependsOnTaskId = null,
|
||||
IReadOnlyList<string>? DependsOnTaskIds = null,
|
||||
bool Blocked = false,
|
||||
string? BlockedReason = null,
|
||||
DirtyBaseWarning? BaseDirty = null,
|
||||
@@ -205,14 +205,14 @@ public sealed class ExternalMcpService
|
||||
}
|
||||
|
||||
// Batch-resolves, for each task, whether the picker is currently skipping it (Queued with
|
||||
// either a planning-chain BlockedByTaskId or an unmet DependsOnTaskId) and why. Only Queued
|
||||
// either a planning-chain BlockedByTaskId or an unmet dependency) and why. Only Queued
|
||||
// tasks can be blocked -- once a task has run, or hasn't been queued yet, blocking is moot.
|
||||
private async Task<Dictionary<string, (bool Blocked, string? Reason)>> ComputeBlockedInfoAsync(
|
||||
IReadOnlyList<TaskEntity> tasks, CancellationToken ct)
|
||||
{
|
||||
var dependencyIds = tasks
|
||||
.Where(t => t.Status == TaskStatus.Queued && t.BlockedByTaskId is null && t.DependsOnTaskId is not null)
|
||||
.Select(t => t.DependsOnTaskId!)
|
||||
.Where(t => t.Status == TaskStatus.Queued && t.BlockedByTaskId is null)
|
||||
.SelectMany(t => t.DependsOnTaskIds)
|
||||
.Distinct()
|
||||
.ToList();
|
||||
|
||||
@@ -237,17 +237,10 @@ public sealed class ExternalMcpService
|
||||
{
|
||||
result[t.Id] = (true, $"Blocked by planning-chain predecessor {t.BlockedByTaskId}.");
|
||||
}
|
||||
else if (t.DependsOnTaskId is not null)
|
||||
{
|
||||
var known = dependencyStatuses.TryGetValue(t.DependsOnTaskId, out var depStatus);
|
||||
result[t.Id] = known && depStatus == TaskStatus.Done
|
||||
? (false, null)
|
||||
: (true, $"Blocked: depends on task {t.DependsOnTaskId} (status: " +
|
||||
(known ? depStatus.ToString() : "not found") + ").");
|
||||
}
|
||||
else
|
||||
{
|
||||
result[t.Id] = (false, null);
|
||||
var unmet = TaskBlocking.Unmet(t.DependsOnTaskIds, dependencyStatuses);
|
||||
result[t.Id] = unmet.Count == 0 ? (false, null) : (true, TaskBlocking.Describe(unmet));
|
||||
}
|
||||
}
|
||||
return result;
|
||||
@@ -298,10 +291,15 @@ public sealed class ExternalMcpService
|
||||
"list/global default (normally sonnet).")]
|
||||
string? model = null,
|
||||
[Description("Id of a task that must reach Done (i.e. be merged) before the picker will claim this one, " +
|
||||
"even once queued. Rejected if it doesn't exist, is this task's own id, or would create a " +
|
||||
"even once queued. Shorthand for a one-element dependsOnTaskIds; ignored when that is given. " +
|
||||
"Rejected if it doesn't exist, is this task's own id, or would create a " +
|
||||
"dependency cycle. If that predecessor instead ends up Failed or Cancelled, this task simply " +
|
||||
"stays blocked rather than starving silently -- check get_task/list_tasks' blocked field.")]
|
||||
string? dependsOnTaskId = null,
|
||||
[Description("Ids of ALL tasks that must reach Done before the picker will claim this one. Use this when a " +
|
||||
"task wires together, or verifies, the results of several independent tasks -- one waiting on " +
|
||||
"A and B is the normal shape of a wave-style plan. Same rejection rules as dependsOnTaskId.")]
|
||||
string[]? dependsOnTaskIds = null,
|
||||
[Description("true: this task's correct outcome may be an empty diff (verification/audit/research work " +
|
||||
"that doesn't change code) — the merge helper won't treat a 0-changed-files result as a red " +
|
||||
"flag. false (default): an empty diff is treated as suspicious, not a clean success.")]
|
||||
@@ -316,7 +314,7 @@ public sealed class ExternalMcpService
|
||||
var list = await _lists.GetByIdAsync(listId, cancellationToken)
|
||||
?? throw new InvalidOperationException($"List {listId} not found.");
|
||||
|
||||
dependsOnTaskId = await TaskIdResolver.ResolveOptionalAsync(_tasks, dependsOnTaskId, cancellationToken);
|
||||
var dependsOn = await ResolveDependsOnAsync(dependsOnTaskId, dependsOnTaskIds, cancellationToken);
|
||||
|
||||
var possibleDuplicates = await FindPossibleDuplicatesAsync(listId, title, cancellationToken);
|
||||
|
||||
@@ -337,12 +335,11 @@ public sealed class ExternalMcpService
|
||||
};
|
||||
await _tasks.AddAsync(entity, cancellationToken);
|
||||
|
||||
if (dependsOnTaskId is not null)
|
||||
if (dependsOn is { Count: > 0 })
|
||||
{
|
||||
var dependsResult = await _state.SetDependsOnAsync(entity.Id, dependsOnTaskId, cancellationToken);
|
||||
var dependsResult = await _state.SetDependsOnAsync(entity.Id, dependsOn, cancellationToken);
|
||||
if (!dependsResult.Ok)
|
||||
throw new InvalidOperationException(dependsResult.Reason ?? "Cannot set dependsOnTaskId.");
|
||||
entity.DependsOnTaskId = dependsOnTaskId;
|
||||
throw new InvalidOperationException(dependsResult.Reason ?? "Cannot set dependsOnTaskIds.");
|
||||
}
|
||||
|
||||
if (queueImmediately)
|
||||
@@ -350,12 +347,39 @@ public sealed class ExternalMcpService
|
||||
var enqueue = await _state.EnqueueAsync(entity.Id, cancellationToken);
|
||||
if (!enqueue.Ok)
|
||||
throw new InvalidOperationException(enqueue.Reason ?? "Cannot enqueue task.");
|
||||
entity.Status = TaskStatus.Queued;
|
||||
}
|
||||
|
||||
await _broadcaster.TaskUpdated(entity.Id);
|
||||
var blocked = await ComputeBlockedInfoAsync([entity], cancellationToken);
|
||||
return new AddTaskResult(ToRefDto(entity, blocked[entity.Id].Blocked, blocked[entity.Id].Reason), possibleDuplicates);
|
||||
// Reload rather than patching `entity`: SetDependsOnAsync and EnqueueAsync both write
|
||||
// through their own context, and adding the edges to this still-tracked entity's nav
|
||||
// would queue a duplicate INSERT on the next SaveChanges.
|
||||
var reload = (await _tasks.GetByIdAsync(entity.Id, cancellationToken))!;
|
||||
var blocked = await ComputeBlockedInfoAsync([reload], cancellationToken);
|
||||
return new AddTaskResult(ToRefDto(reload, blocked[reload.Id].Blocked, blocked[reload.Id].Reason), possibleDuplicates);
|
||||
}
|
||||
|
||||
// Folds add_task/update_task's two dependency parameters into one resolved set.
|
||||
// `dependsOnTaskIds` wins when given; otherwise the legacy scalar form is treated as a
|
||||
// one-element set, with the empty string still its documented "clear it" sentinel.
|
||||
// Returns null only when neither parameter was passed = leave the task's set untouched.
|
||||
private async Task<IReadOnlyList<string>?> ResolveDependsOnAsync(
|
||||
string? dependsOnTaskId, string[]? dependsOnTaskIds, CancellationToken ct)
|
||||
{
|
||||
if (dependsOnTaskIds is not null)
|
||||
{
|
||||
var resolved = new List<string>();
|
||||
foreach (var raw in dependsOnTaskIds)
|
||||
{
|
||||
var id = await TaskIdResolver.ResolveOptionalAsync(_tasks, raw, ct);
|
||||
if (id.NullIfBlank() is { } v) resolved.Add(v);
|
||||
}
|
||||
return resolved;
|
||||
}
|
||||
|
||||
if (dependsOnTaskId is null) return null;
|
||||
// TaskIdResolver passes "" through untouched, so NullIfBlank still yields the clear signal.
|
||||
var single = (await TaskIdResolver.ResolveOptionalAsync(_tasks, dependsOnTaskId, ct)).NullIfBlank();
|
||||
return single is null ? [] : [single];
|
||||
}
|
||||
|
||||
// Non-terminal: a task still open enough that a new, similarly-titled task might be a duplicate
|
||||
@@ -436,10 +460,15 @@ public sealed class ExternalMcpService
|
||||
string? title = null,
|
||||
string? description = null,
|
||||
string? commitType = null,
|
||||
[Description("Id of a task that must reach Done before the picker will claim this one. Pass an empty " +
|
||||
"string to clear an existing link; null leaves it unchanged. Rejected if it doesn't exist, " +
|
||||
[Description("Id of a task that must reach Done before the picker will claim this one. Shorthand for a " +
|
||||
"one-element dependsOnTaskIds; ignored when that is given. Pass an empty " +
|
||||
"string to clear all existing links; null leaves them unchanged. Rejected if it doesn't exist, " +
|
||||
"is this task's own id, or would create a dependency cycle.")]
|
||||
string? dependsOnTaskId = null,
|
||||
[Description("Ids of ALL tasks that must reach Done before the picker will claim this one, replacing the " +
|
||||
"task's current set. Pass an empty array to clear it; null leaves it unchanged. Use this when " +
|
||||
"a task wires together, or verifies, the results of several independent tasks.")]
|
||||
string[]? dependsOnTaskIds = null,
|
||||
[Description("true: mark this task as a manual reminder only the user can complete — the queue picker, " +
|
||||
"daily prep, and the list handler all skip it (a ConPTY session may still touch it). false: " +
|
||||
"clear the flag. null (default): leave unchanged.")]
|
||||
@@ -463,15 +492,12 @@ public sealed class ExternalMcpService
|
||||
if (expectsNoDiff is not null) task.ExpectsNoDiff = expectsNoDiff.Value;
|
||||
await _tasks.UpdateAsync(task, cancellationToken);
|
||||
|
||||
if (dependsOnTaskId is not null)
|
||||
var dependsOn = await ResolveDependsOnAsync(dependsOnTaskId, dependsOnTaskIds, cancellationToken);
|
||||
if (dependsOn is not null)
|
||||
{
|
||||
// Empty string is a deliberate sentinel (see the dependsOnTaskId parameter doc) to
|
||||
// clear the link — TaskIdResolver passes it through untouched, then NullIfBlank
|
||||
// below still turns it into the clear signal SetDependsOnAsync expects.
|
||||
dependsOnTaskId = await TaskIdResolver.ResolveOptionalAsync(_tasks, dependsOnTaskId, cancellationToken);
|
||||
var dependsResult = await _state.SetDependsOnAsync(taskId, dependsOnTaskId.NullIfBlank(), cancellationToken);
|
||||
var dependsResult = await _state.SetDependsOnAsync(taskId, dependsOn, cancellationToken);
|
||||
if (!dependsResult.Ok)
|
||||
throw new InvalidOperationException(dependsResult.Reason ?? "Cannot set dependsOnTaskId.");
|
||||
throw new InvalidOperationException(dependsResult.Reason ?? "Cannot set dependsOnTaskIds.");
|
||||
}
|
||||
|
||||
var reload = (await _tasks.GetByIdAsync(taskId, cancellationToken))!;
|
||||
@@ -742,6 +768,7 @@ public sealed class ExternalMcpService
|
||||
var children = await ctx.Tasks
|
||||
.AsNoTracking()
|
||||
.Include(t => t.Worktree)
|
||||
.Include(t => t.Dependencies)
|
||||
.Where(t => t.ParentTaskId == parentTaskId && t.Status == TaskStatus.Done)
|
||||
.ToListAsync(ct);
|
||||
|
||||
@@ -935,7 +962,8 @@ public sealed class ExternalMcpService
|
||||
|
||||
await using var ctx = await _dbFactory.CreateDbContextAsync(cancellationToken);
|
||||
|
||||
var task = await ctx.Tasks.FirstOrDefaultAsync(t => t.Id == taskId, cancellationToken)
|
||||
var task = await ctx.Tasks.Include(t => t.Dependencies)
|
||||
.FirstOrDefaultAsync(t => t.Id == taskId, cancellationToken)
|
||||
?? throw new InvalidOperationException($"Task {taskId} not found.");
|
||||
|
||||
if (isMyDay && !task.IsMyDay)
|
||||
@@ -995,7 +1023,7 @@ public sealed class ExternalMcpService
|
||||
FailureReasonOf(t),
|
||||
t.Status == TaskStatus.Failed ? t.FailureTurnsUsed : null,
|
||||
t.Status == TaskStatus.Failed ? t.FailureMaxTurns : null,
|
||||
t.DependsOnTaskId,
|
||||
t.DependsOnTaskIds,
|
||||
blocked,
|
||||
blockedReason,
|
||||
t.IsManual,
|
||||
@@ -1014,7 +1042,7 @@ public sealed class ExternalMcpService
|
||||
FailureReasonOf(t),
|
||||
t.Status == TaskStatus.Failed ? t.FailureTurnsUsed : null,
|
||||
t.Status == TaskStatus.Failed ? t.FailureMaxTurns : null,
|
||||
t.DependsOnTaskId,
|
||||
t.DependsOnTaskIds,
|
||||
blocked,
|
||||
blockedReason,
|
||||
BaseDirty: null,
|
||||
|
||||
+42
@@ -0,0 +1,42 @@
|
||||
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
|
||||
|
||||
namespace ClaudeDo.Worker.External;
|
||||
|
||||
/// <summary>
|
||||
/// Renders the `blockedReason` string shared by <see cref="ExternalMcpService"/> and
|
||||
/// <see cref="TaskWaitMcpTools"/>. Lives here so the two can't drift on the wording an agent
|
||||
/// parses.
|
||||
/// </summary>
|
||||
internal static class TaskBlocking
|
||||
{
|
||||
/// <summary>
|
||||
/// The predecessors of <paramref name="dependencyIds"/> that have not reached Done — an id
|
||||
/// missing from <paramref name="statuses"/> counts as unmet (the picker treats a dangling
|
||||
/// edge as blocking too).
|
||||
/// </summary>
|
||||
public static List<(string Id, TaskStatus? Status)> Unmet(
|
||||
IEnumerable<string> dependencyIds, IReadOnlyDictionary<string, TaskStatus> statuses)
|
||||
{
|
||||
var unmet = new List<(string Id, TaskStatus? Status)>();
|
||||
foreach (var id in dependencyIds)
|
||||
{
|
||||
if (statuses.TryGetValue(id, out var status))
|
||||
{
|
||||
if (status != TaskStatus.Done) unmet.Add((id, status));
|
||||
}
|
||||
else
|
||||
{
|
||||
unmet.Add((id, null));
|
||||
}
|
||||
}
|
||||
return unmet;
|
||||
}
|
||||
|
||||
/// <summary>Names every open predecessor, not just the first — a task can wait on n of them.</summary>
|
||||
public static string Describe(IReadOnlyList<(string Id, TaskStatus? Status)> unmet)
|
||||
{
|
||||
var noun = unmet.Count == 1 ? "task" : "tasks";
|
||||
var parts = unmet.Select(u => $"{u.Id} (status: {(u.Status?.ToString() ?? "not found")})");
|
||||
return $"Blocked: depends on {noun} {string.Join(", ", parts)}.";
|
||||
}
|
||||
}
|
||||
+16
-13
@@ -132,17 +132,24 @@ public sealed class TaskWaitMcpTools
|
||||
var rows = await ctx.Tasks
|
||||
.AsNoTracking()
|
||||
.Where(t => taskIds.Contains(t.Id))
|
||||
.Select(t => new { t.Id, t.Number, t.Status, t.BlockedByTaskId, t.DependsOnTaskId })
|
||||
.Select(t => new
|
||||
{
|
||||
t.Id,
|
||||
t.Number,
|
||||
t.Status,
|
||||
t.BlockedByTaskId,
|
||||
DependsOnTaskIds = t.Dependencies.Select(d => d.DependsOnTaskId).ToList(),
|
||||
})
|
||||
.ToListAsync(ct);
|
||||
|
||||
var byId = rows.ToDictionary(r => r.Id, r => r);
|
||||
|
||||
// A Queued task with an unmet depends_on link never becomes "changed" by itself -- the
|
||||
// A Queued task with an unmet depends_on edge never becomes "changed" by itself -- the
|
||||
// picker will not touch it. Batch-resolve those dependencies' statuses once instead of a
|
||||
// query per candidate.
|
||||
var dependencyIds = rows
|
||||
.Where(r => r.Status == TaskStatus.Queued && r.BlockedByTaskId is null && r.DependsOnTaskId is not null)
|
||||
.Select(r => r.DependsOnTaskId!)
|
||||
.Where(r => r.Status == TaskStatus.Queued && r.BlockedByTaskId is null)
|
||||
.SelectMany(r => r.DependsOnTaskIds)
|
||||
.Distinct()
|
||||
.ToList();
|
||||
var dependencyStatuses = dependencyIds.Count == 0
|
||||
@@ -168,16 +175,12 @@ public sealed class TaskWaitMcpTools
|
||||
// every chained child and turn the wait into a one-turn-per-poll busy loop.
|
||||
if (row.Status == TaskStatus.Queued && row.BlockedByTaskId is null)
|
||||
{
|
||||
if (row.DependsOnTaskId is not null)
|
||||
var unmet = TaskBlocking.Unmet(row.DependsOnTaskIds, dependencyStatuses);
|
||||
if (unmet.Count > 0)
|
||||
{
|
||||
var known = dependencyStatuses.TryGetValue(row.DependsOnTaskId, out var depStatus);
|
||||
if (!known || depStatus != TaskStatus.Done)
|
||||
{
|
||||
result.Add(new TaskStatusChangeDto(id, "Blocked",
|
||||
$"Blocked: depends on task {row.DependsOnTaskId} (status: " +
|
||||
(known ? depStatus.ToString() : "not found") + ").", row.Number));
|
||||
continue;
|
||||
}
|
||||
result.Add(new TaskStatusChangeDto(
|
||||
id, "Blocked", TaskBlocking.Describe(unmet), row.Number));
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -21,9 +21,12 @@ public sealed class QueuePicker : IQueuePicker
|
||||
// Atomic queue claim: UPDATE + RETURNING in a single statement prevents TOCTOU races.
|
||||
// Raw SQL because EF cannot express UPDATE...RETURNING.
|
||||
// Eligible task must be Queued, unblocked (chain and depends-on), not manual, and due
|
||||
// (or unscheduled). depends_on_task_id only gates on the dependency's Status='done' --
|
||||
// unlike blocked_by_task_id there is no cascade on the dependency failing, so a task
|
||||
// with a Failed dependency simply stays here, skipped, until someone intervenes.
|
||||
// (or unscheduled). A task_dependencies edge only gates on the dependency's Status='done'
|
||||
// -- unlike blocked_by_task_id there is no cascade on the dependency failing, so a task
|
||||
// with a Failed dependency simply stays here, skipped, until someone intervenes. The
|
||||
// depends-on clause is doubly negated on purpose: "no edge whose predecessor isn't done"
|
||||
// means a task with zero edges is eligible, ALL edges must be done, and an edge pointing
|
||||
// at a vanished task keeps blocking (same as the old single-column form did).
|
||||
// EF SQLite stores DateTime as "yyyy-MM-dd HH:mm:ss.fffffff" — same format used here for comparison.
|
||||
var rows = await ctx.Tasks.FromSqlRaw("""
|
||||
UPDATE tasks SET status = 'running', started_at = {1}
|
||||
@@ -33,11 +36,14 @@ public sealed class QueuePicker : IQueuePicker
|
||||
AND t.blocked_by_task_id IS NULL
|
||||
AND t.is_manual = 0
|
||||
AND (t.scheduled_for IS NULL OR t.scheduled_for <= {0})
|
||||
AND (t.depends_on_task_id IS NULL
|
||||
OR EXISTS (
|
||||
SELECT 1 FROM tasks d
|
||||
WHERE d.id = t.depends_on_task_id AND d.status = 'done'
|
||||
))
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM task_dependencies td
|
||||
WHERE td.task_id = t.id
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM tasks d
|
||||
WHERE d.id = td.depends_on_task_id AND d.status = 'done'
|
||||
)
|
||||
)
|
||||
ORDER BY t.sort_order ASC, t.created_at ASC
|
||||
LIMIT 1
|
||||
)
|
||||
|
||||
@@ -37,9 +37,10 @@ public interface ITaskStateService
|
||||
Task<TransitionResult> BlockOnAsync(string taskId, string predecessorTaskId, CancellationToken ct);
|
||||
Task<TransitionResult> UnblockAsync(string taskId, CancellationToken ct);
|
||||
|
||||
// dependsOnTaskId null clears the dependency. Rejects self-reference, an unknown dependency
|
||||
// id, and a link that would create a cycle -- see TaskStateService for the walk.
|
||||
Task<TransitionResult> SetDependsOnAsync(string taskId, string? dependsOnTaskId, CancellationToken ct);
|
||||
// Replaces the task's whole dependency set; an empty list clears it. Rejects a self-reference,
|
||||
// an unknown dependency id, and any link that would create a cycle in the dependency graph
|
||||
// -- see TaskStateService for the walk.
|
||||
Task<TransitionResult> SetDependsOnAsync(string taskId, IReadOnlyList<string> dependsOnTaskIds, CancellationToken ct);
|
||||
|
||||
// Surfaces a WaitingForChildren parent for review once all its children are terminal.
|
||||
// Best-effort (swallows and logs failures) — safe to call after any child mutation,
|
||||
|
||||
@@ -501,50 +501,89 @@ public sealed class TaskStateService : ITaskStateService
|
||||
return new TransitionResult(true, null);
|
||||
}
|
||||
|
||||
public async Task<TransitionResult> SetDependsOnAsync(string taskId, string? dependsOnTaskId, CancellationToken ct)
|
||||
public async Task<TransitionResult> SetDependsOnAsync(
|
||||
string taskId, IReadOnlyList<string> dependsOnTaskIds, CancellationToken ct)
|
||||
{
|
||||
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
||||
|
||||
if (dependsOnTaskId is not null)
|
||||
{
|
||||
if (dependsOnTaskId == taskId)
|
||||
return new TransitionResult(false, "A task cannot depend on itself.");
|
||||
|
||||
if (!await ctx.Tasks.AsNoTracking().AnyAsync(t => t.Id == dependsOnTaskId, ct))
|
||||
return new TransitionResult(false, $"Dependency task {dependsOnTaskId} not found.");
|
||||
|
||||
// Walk the proposed predecessor's own chain of dependencies; if it leads back to
|
||||
// taskId, linking here would create a cycle that starves both tasks forever (the
|
||||
// picker never claims either). `visited` also stops us looping forever on
|
||||
// pre-existing bad data unrelated to this write.
|
||||
var current = dependsOnTaskId;
|
||||
var visited = new HashSet<string>();
|
||||
while (current is not null)
|
||||
{
|
||||
if (current == taskId)
|
||||
return new TransitionResult(false, "Setting this dependency would create a cycle.");
|
||||
if (!visited.Add(current))
|
||||
break;
|
||||
current = await ctx.Tasks.AsNoTracking()
|
||||
.Where(t => t.Id == current)
|
||||
.Select(t => t.DependsOnTaskId)
|
||||
.FirstOrDefaultAsync(ct);
|
||||
}
|
||||
}
|
||||
|
||||
var affected = await ctx.Tasks
|
||||
.Where(t => t.Id == taskId)
|
||||
.ExecuteUpdateAsync(s => s.SetProperty(t => t.DependsOnTaskId, dependsOnTaskId), ct);
|
||||
|
||||
if (affected == 0)
|
||||
if (!await ctx.Tasks.AsNoTracking().AnyAsync(t => t.Id == taskId, ct))
|
||||
return new TransitionResult(false, "Task not found.");
|
||||
|
||||
// Clearing a dependency may free up a Queued task the picker was skipping.
|
||||
if (dependsOnTaskId is null) _waker.Wake();
|
||||
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;
|
||||
|
||||
Reference in New Issue
Block a user