feat(worker): fast-forward finished children into the parent's unit branch

Planning parents get their own worktree on Queue plan, children fork from the
unit head and are absorbed via merge --ff-only before Done (ff failure -> Failed,
interactive child approve -> absorb instead of merge to the target).
This commit is contained in:
mika kuns
2026-10-01 11:25:47 +02:00
parent 724ad90f88
commit 9da522b0c9
8 changed files with 489 additions and 17 deletions
@@ -117,6 +117,7 @@ public sealed class TaskMergeService
private readonly ITaskStateService _state;
private readonly IVerifyCommandRunner _verify;
private readonly ILogger<TaskMergeService> _logger;
private readonly UnitBranchService? _unitBranch;
public TaskMergeService(
IDbContextFactory<ClaudeDoDbContext> dbFactory,
@@ -124,7 +125,8 @@ public sealed class TaskMergeService
HubBroadcaster broadcaster,
ITaskStateService state,
IVerifyCommandRunner verify,
ILogger<TaskMergeService> logger)
ILogger<TaskMergeService> logger,
UnitBranchService? unitBranch = null)
{
_dbFactory = dbFactory;
_git = git;
@@ -132,6 +134,7 @@ public sealed class TaskMergeService
_state = state;
_verify = verify;
_logger = logger;
_unitBranch = unitBranch;
}
private async Task<(TaskEntity Task, ListEntity List, WorktreeEntity? Worktree, string? VerifyCommand)> LoadMergeContextAsync(
@@ -1010,6 +1013,21 @@ public sealed class TaskMergeService
if (task.Status != TaskStatus.WaitingForReview)
return Blocked("task is not waiting for review");
// An interactive child of a unit lands in the unit branch, not on the target; the parent's
// own approve merges the unit. A failed fast-forward keeps the child in review.
if (task.ParentTaskId is not null && _unitBranch is not null)
{
var absorb = await _unitBranch.AbsorbChildAsync(taskId, ct);
if (absorb.Failed) return Blocked(absorb.Error!);
if (absorb.Applicable)
{
var approved = await _state.ApproveReviewAsync(taskId, ct);
return approved.Ok
? new MergeResult(StatusMerged, Array.Empty<string>(), null)
: Blocked(approved.Reason ?? "approve failed");
}
}
if (wt is not null && wt.State == WorktreeState.Kept)
{
// Kept means either RevertMergeAsync undid a previous merge, or the user manually
@@ -1,5 +1,6 @@
using ClaudeDo.Data;
using ClaudeDo.Data.Models;
using ClaudeDo.Worker.Runner;
using ClaudeDo.Worker.State;
using Microsoft.EntityFrameworkCore;
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
@@ -10,13 +11,16 @@ public sealed class PlanningChainCoordinator
{
private readonly IDbContextFactory<ClaudeDoDbContext> _dbFactory;
private readonly Func<ITaskStateService> _state;
private readonly UnitBranchService? _unitBranch;
public PlanningChainCoordinator(
IDbContextFactory<ClaudeDoDbContext> dbFactory,
Func<ITaskStateService> state)
Func<ITaskStateService> state,
UnitBranchService? unitBranch = null)
{
_dbFactory = dbFactory;
_state = state;
_unitBranch = unitBranch;
}
// Sets up a sequential chain over a planning parent's children.
@@ -86,6 +90,9 @@ public sealed class PlanningChainCoordinator
if (phase != PlanningPhase.Finalized)
throw new InvalidOperationException("Plan must be finalized before it can be queued.");
// The parent's worktree is the unit branch every child forks from and is absorbed into.
if (_unitBranch is not null)
await _unitBranch.EnsureUnitWorktreeAsync(parentTaskId, ct);
return await SetupChainAsync(parentTaskId, enqueue: true, ct);
}
+1
View File
@@ -112,6 +112,7 @@ builder.Services.AddSingleton<HubBroadcaster>();
builder.Services.AddSingleton<ConnectedClientTracker>();
builder.Services.AddSingleton<GitService>();
builder.Services.AddSingleton<WorktreeManager>();
builder.Services.AddSingleton<UnitBranchService>();
builder.Services.AddSingleton<ClaudeArgsBuilder>();
builder.Services.AddSingleton<TaskRunTokenRegistry>();
builder.Services.AddSingleton<PendingQuestionRegistry>();
+14 -1
View File
@@ -28,6 +28,7 @@ public sealed class TaskRunner
private readonly AttachmentStore _attachments;
private readonly ISessionSkillSeeder _skillSeeder;
private readonly ITranscriptUsageReader _usageReader;
private readonly UnitBranchService? _unitBranch;
public TaskRunner(
IClaudeProcess claude,
@@ -41,7 +42,8 @@ public sealed class TaskRunner
TaskRunTokenRegistry tokens,
AttachmentStore attachments,
ISessionSkillSeeder skillSeeder,
ITranscriptUsageReader usageReader)
ITranscriptUsageReader usageReader,
UnitBranchService? unitBranch = null)
{
_claude = claude;
_dbFactory = dbFactory;
@@ -55,6 +57,7 @@ public sealed class TaskRunner
_attachments = attachments;
_skillSeeder = skillSeeder;
_usageReader = usageReader;
_unitBranch = unitBranch;
}
public async Task RunAsync(TaskEntity task, string slot, CancellationToken ct, bool alreadyClaimed = false)
@@ -568,6 +571,16 @@ public sealed class TaskRunner
}
else
{
// Land the child in its unit branch before Done unblocks the successor, which forks from that head.
if (_unitBranch is not null)
{
var absorb = await _unitBranch.AbsorbChildAsync(task.Id, CancellationToken.None);
if (absorb.Failed)
{
await MarkFailed(task.Id, task.Number, task.Title, slot, absorb.Error, result.TurnCount);
return;
}
}
var transition = await _state.CompleteAsync(task.Id, finishedAt, reviewResult, CancellationToken.None);
if (!transition.Ok)
_logger.LogWarning("CompleteAsync rejected for task {TaskId}: {Reason}", task.Id, transition.Reason);
@@ -0,0 +1,122 @@
using ClaudeDo.Data;
using ClaudeDo.Data.Git;
using ClaudeDo.Data.Models;
using ClaudeDo.Data.Repositories;
using ClaudeDo.Worker.Hub;
using Microsoft.EntityFrameworkCore;
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
namespace ClaudeDo.Worker.Runner;
/// <summary>Outcome of <see cref="UnitBranchService.AbsorbChildAsync"/>. <c>Applicable</c> = the child belongs to a unit branch.</summary>
public sealed record AbsorbResult(bool Applicable, string? Error)
{
/// <summary>The child has no unit branch to land in (no parent worktree, or no own worktree).</summary>
public static readonly AbsorbResult NotApplicable = new(false, null);
/// <summary>The unit branch was fast-forwarded to the child.</summary>
public static readonly AbsorbResult Absorbed = new(true, null);
/// <summary>True when the fast-forward was impossible; <see cref="Error"/> says why.</summary>
public bool Failed => Error is not null;
}
/// <summary>
/// Git side of the unit branch: the parent's worktree owns it, children are fast-forwarded into it.
/// Never writes task status — callers own the transitions, which keeps the DI graph acyclic.
/// </summary>
public sealed class UnitBranchService
{
private readonly GitService _git;
private readonly WorktreeManager _worktrees;
private readonly IDbContextFactory<ClaudeDoDbContext> _dbFactory;
private readonly ILogger<UnitBranchService> _logger;
private readonly HubBroadcaster? _broadcaster;
public UnitBranchService(
GitService git, WorktreeManager worktrees, IDbContextFactory<ClaudeDoDbContext> dbFactory,
ILogger<UnitBranchService> logger, HubBroadcaster? broadcaster = null)
{
_git = git;
_worktrees = worktrees;
_dbFactory = dbFactory;
_logger = logger;
_broadcaster = broadcaster;
}
/// <summary>Gives a parent its own worktree (= unit branch) unless it already has an Active one or its list has no working dir.</summary>
public async Task EnsureUnitWorktreeAsync(string parentId, CancellationToken ct)
{
TaskEntity parent;
ListEntity list;
using (var ctx = _dbFactory.CreateDbContext())
{
parent = await ctx.Tasks.AsNoTracking().FirstOrDefaultAsync(t => t.Id == parentId, ct)
?? throw new KeyNotFoundException($"Task '{parentId}' not found.");
var wt = await new WorktreeRepository(ctx).GetByTaskIdAsync(parentId, ct);
if (wt is { State: WorktreeState.Active }) return;
list = await new ListRepository(ctx).GetByIdAsync(parent.ListId, ct)
?? throw new InvalidOperationException("List not found.");
}
if (string.IsNullOrWhiteSpace(list.WorkingDir)) return;
await _worktrees.CreateAsync(parent, list, ct);
if (_broadcaster is not null) await _broadcaster.WorktreeUpdated(parentId);
}
/// <summary>
/// Fast-forwards the parent's unit branch to the child's branch, then marks the child row
/// <c>Merged</c> (no merge commit) and removes its worktree and branch.
/// </summary>
public async Task<AbsorbResult> AbsorbChildAsync(string childId, CancellationToken ct)
{
WorktreeEntity? childWt, parentWt;
string? workingDir;
using (var ctx = _dbFactory.CreateDbContext())
{
var child = await ctx.Tasks.AsNoTracking().Include(t => t.List)
.FirstOrDefaultAsync(t => t.Id == childId, ct);
if (child?.ParentTaskId is null) return AbsorbResult.NotApplicable;
var repo = new WorktreeRepository(ctx);
childWt = await repo.GetByTaskIdAsync(childId, ct);
parentWt = await repo.GetByTaskIdAsync(child.ParentTaskId, ct);
workingDir = child.List.WorkingDir;
}
if (childWt is not { State: WorktreeState.Active } || parentWt is not { State: WorktreeState.Active }
|| string.IsNullOrWhiteSpace(workingDir))
return AbsorbResult.NotApplicable;
// ponytail: kein Merge-Fallback; einbauen, falls parallele Children ein echter Fall werden.
var (code, stderr) = await _git.MergeFastForwardOnlyAsync(parentWt.Path, childWt.BranchName, ct);
if (code != 0)
return new AbsorbResult(true,
$"Unit branch '{parentWt.BranchName}' cannot be fast-forwarded to '{childWt.BranchName}' — the child " +
$"did not start from the unit's current head (e.g. it was run in parallel via Run now). git: {stderr}");
var head = await _git.RevParseHeadAsync(parentWt.Path, ct);
var diffStat = await _git.DiffStatAsync(parentWt.Path, parentWt.BaseCommit, head, ct: ct);
using (var ctx = _dbFactory.CreateDbContext())
{
var repo = new WorktreeRepository(ctx);
await repo.UpdateHeadAsync(parentWt.TaskId, head, diffStat, ct);
await repo.SetMergedAsync(childId, null, ct);
}
// The work is in the unit branch now; a leftover folder or branch is only clutter.
try { await _git.WorktreeRemoveAsync(workingDir, childWt.Path, force: true, CancellationToken.None); }
catch (Exception ex) { _logger.LogWarning(ex, "Failed to remove absorbed worktree {Path}", childWt.Path); }
try { await _git.BranchDeleteAsync(workingDir, childWt.BranchName, force: true, CancellationToken.None); }
catch (Exception ex) { _logger.LogWarning(ex, "Failed to delete absorbed branch {Branch}", childWt.BranchName); }
_logger.LogInformation("Absorbed {Child} into unit branch {Branch} at {Head}", childWt.BranchName, parentWt.BranchName, head);
if (_broadcaster is not null)
{
await _broadcaster.WorktreeUpdated(childId);
await _broadcaster.WorktreeUpdated(parentWt.TaskId);
}
return AbsorbResult.Absorbed;
}
/// <summary>Legacy unit: a Done child whose worktree is still Active was never absorbed and needs the unit merge.</summary>
public static Task<bool> HasUnabsorbedChildrenAsync(ClaudeDoDbContext ctx, string parentId, CancellationToken ct) =>
ctx.Tasks.AnyAsync(t => t.ParentTaskId == parentId && t.Status == TaskStatus.Done
&& t.Worktree != null && t.Worktree.State == WorktreeState.Active, ct);
}
+8 -12
View File
@@ -256,23 +256,19 @@ public sealed class WorktreeManager
return TicketStatusMap.ParseBandelTicketId(parentRef);
}
// Improvement children (parent is a non-planning task with its own worktree) branch
// from the parent's recorded HEAD so they build on the parent's not-yet-merged work.
// Planning children and standalone tasks base off list_config.base_branch, else the current HEAD.
/// <summary>
/// A child of a parent with an Active worktree starts from the unit branch's current head, so it
/// sees every sibling absorbed before it. Unit roots (and legacy children) base off
/// <c>list_config.base_branch</c>, else the current HEAD.
/// </summary>
private async Task<string> ResolveBaseCommitAsync(TaskEntity task, string workingDir, CancellationToken ct)
{
if (task.ParentTaskId is not null)
{
using var ctx = _dbFactory.CreateDbContext();
var parent = await ctx.Tasks.AsNoTracking()
.FirstOrDefaultAsync(t => t.Id == task.ParentTaskId, ct);
if (parent is not null && parent.PlanningPhase == PlanningPhase.None)
{
var parentWt = await new WorktreeRepository(ctx).GetByTaskIdAsync(task.ParentTaskId, ct);
var parentHead = parentWt?.HeadCommit ?? parentWt?.BaseCommit;
if (parentHead is not null)
return parentHead;
}
var parentWt = await new WorktreeRepository(ctx).GetByTaskIdAsync(task.ParentTaskId, ct);
if (parentWt is { State: WorktreeState.Active })
return await _git.RevParseAsync(workingDir, parentWt.BranchName, ct);
}
using var cfgCtx = _dbFactory.CreateDbContext();
var baseBranch = await cfgCtx.ListConfigs.AsNoTracking()