feat(worker-mcp): raise wait_for_task_change timeout, expose queue slot state

MaxTimeoutSeconds was 170s against runs that take tens of minutes, forcing
a dozen full-context wait rounds per long-running batch. Raise it to 900s
and raise MCP_TOOL_TIMEOUT in lockstep (ClaudeProcess + every
InteractiveLaunchSpecService launch spec) to 930000ms so the client
connection actually stays open that long instead of aborting first.

Add get_queue_state (QueueStateMcpTools): configured vs. effective
parallel-slot count (via QueueService.GetSlotCountsAsync, extracted from
the former GetEffectiveMaxParallelAsync), active slots with taskId +
startedAt including the run_task_now override slot, and queued tasks in
pick order -- so a caller can observe queue occupancy instead of inferring
it from maxParallelExecutions.
This commit is contained in:
mika kuns
2026-08-05 20:47:57 +02:00
parent bdee731376
commit d43b5fcefc
10 changed files with 299 additions and 36 deletions
+24 -6
View File
@@ -1,8 +1,8 @@
# External MCP tool surface # External MCP tool surface
> **Explore-note — verify before trusting.** Distilled map of a subsystem, not authoritative. > **Explore-note — verify before trusting.** Distilled map of a subsystem, not authoritative.
> Last verified against commit `f6cb825` (2026-08-05). > Last verified against commit `bdee731` (2026-08-05).
> Drift check: `git log --oneline f6cb825..HEAD -- src/ClaudeDo.Worker/External` > Drift check: `git log --oneline bdee731..HEAD -- src/ClaudeDo.Worker/External`
> Stable structure only (no line numbers). See docs/explore-notes/README.md. > Stable structure only (no line numbers). See docs/explore-notes/README.md.
Covers `src/ClaudeDo.Worker/External/` — the always-on MCP tools ClaudeDo exposes to general Covers `src/ClaudeDo.Worker/External/` — the always-on MCP tools ClaudeDo exposes to general
@@ -57,6 +57,7 @@ Daily prep: `GetDailyPrepCandidates`, `SetMyDay`.
| `LifecycleMcpTools` | `ResetFailedTask` | | `LifecycleMcpTools` | `ResetFailedTask` |
| `AppSettingsMcpTools` | `GetAppSettings` (read-only) | | `AppSettingsMcpTools` | `GetAppSettings` (read-only) |
| `TaskWaitMcpTools` | `WaitForTaskChange` | | `TaskWaitMcpTools` | `WaitForTaskChange` |
| `QueueStateMcpTools` | `GetQueueState` |
| `AttachmentMcpTools` | `AddTaskAttachment`, `ListTaskAttachments`, `RemoveTaskAttachment` | | `AttachmentMcpTools` | `AddTaskAttachment`, `ListTaskAttachments`, `RemoveTaskAttachment` |
## Per-tool behaviour worth knowing ## Per-tool behaviour worth knowing
@@ -110,12 +111,29 @@ failing item never aborts the rest — and rejects batches over **100 items**.
- Implemented as an **async DB poll** (short-lived `DbContext` per check, 500 ms delay, no - Implemented as an **async DB poll** (short-lived `DbContext` per check, 500 ms delay, no
held connection, no busy loop) rather than hooking `HubBroadcaster` — deliberately isolated held connection, no busy loop) rather than hooking `HubBroadcaster` — deliberately isolated
so it can't regress the existing broadcast callers. so it can't regress the existing broadcast callers.
- `timeoutSeconds` is clamped server-side to `TaskWaitMcpTools.MaxTimeoutSeconds` (170 s), - `timeoutSeconds` is clamped server-side to `TaskWaitMcpTools.MaxTimeoutSeconds` (900 s),
comfortably under the list handler's `MCP_TOOL_TIMEOUT` (200 s, see comfortably under the `MCP_TOOL_TIMEOUT` (930 s) every ClaudeDo-owned claude launcher sets —
`InteractiveLaunchSpecService`), so the tool reports `timedOut: true` instead of racing the `ClaudeProcess` for headless queue runs, `InteractiveLaunchSpecService` for every embedded
client's own abort. ConPTY session (list handler, planning, interactive resume) — so the tool reports
`timedOut: true` instead of racing the client's own abort. A caller running claude with a
different (or default: 60 s) `MCP_TOOL_TIMEOUT` will still see its own client-side timeout
fire first; the server has no way to detect or compensate for that.
- Replaced the list handler's old "sleep + poll `get_task` in a loop" Phase 3 instruction. - Replaced the list handler's old "sleep + poll `get_task` in a loop" Phase 3 instruction.
**`GetQueueState()`** — read-only snapshot so a caller doesn't have to infer queue state from
`maxParallelExecutions` or repeated `wait_for_task_change` rounds:
`{ configuredSlots, effectiveSlots, activeSlots: [{ slot, taskId, startedAt }], waitingTaskIds }`.
- `configuredSlots`/`effectiveSlots` reuse `QueueService.GetSlotCountsAsync` — the same
configured-vs-throttled computation `QueueService.ExecuteAsync` uses each tick (see
[usage-monitoring](usage-monitoring.md) for the throttle staging) — so this tool can't drift
from the queue's actual refill decision.
- `activeSlots` reuses `QueueService.GetActive()` (already the source for the Hub's `GetActive`):
`slot` is `"queue"` for a normal queue slot or `"override"` for the single
`run_task_now`/`continue_task` slot.
- `waitingTaskIds` is a fresh read-only query mirroring `QueuePicker.ClaimNextAsync`'s
eligibility filter and order (`Queued`, unblocked, non-manual, due, `sort_order` then
`created_at`) — it does not claim or mutate anything.
**`AttachmentMcpTools`** — re-attaching the same `fileName` overwrites. Add/remove refuse on a **`AttachmentMcpTools`** — re-attaching the same `fileName` overwrites. Add/remove refuse on a
`Running` task. `Running` task.
+61
View File
@@ -0,0 +1,61 @@
using System.ComponentModel;
using ClaudeDo.Data;
using ClaudeDo.Worker.Queue;
using Microsoft.EntityFrameworkCore;
using ModelContextProtocol.Server;
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
namespace ClaudeDo.Worker.External;
public sealed record QueueSlotDto(string Slot, string TaskId, DateTime StartedAt);
public sealed record GetQueueStateResult(
int ConfiguredSlots,
int EffectiveSlots,
IReadOnlyList<QueueSlotDto> ActiveSlots,
IReadOnlyList<string> WaitingTaskIds);
[McpServerToolType]
public sealed class QueueStateMcpTools
{
private readonly QueueService _queue;
private readonly IDbContextFactory<ClaudeDoDbContext> _dbFactory;
public QueueStateMcpTools(QueueService queue, IDbContextFactory<ClaudeDoDbContext> dbFactory)
{
_queue = queue;
_dbFactory = dbFactory;
}
[McpServerTool, Description(
"Read-only snapshot of the execution queue -- observe slot occupancy instead of inferring " +
"it from maxParallelExecutions. Result: { configuredSlots, effectiveSlots, activeSlots: " +
"[{ slot, taskId, startedAt }], waitingTaskIds }. configuredSlots is Settings -> " +
"MaxParallelExecutions; effectiveSlots is that value stepped down by the usage throttle " +
"(lower when the 5h/7d usage window is filling up) -- compare the two to see whether " +
"throttling is currently active. activeSlots lists every task presently holding an " +
"execution slot, with slot \"queue\" for a normal queue slot or \"override\" for the single " +
"run_task_now/continue_task slot. waitingTaskIds lists queued, unblocked, non-manual, due " +
"tasks in the order the queue would pick them next.")]
public async Task<GetQueueStateResult> GetQueueState(CancellationToken cancellationToken = default)
{
var (configured, effective) = await _queue.GetSlotCountsAsync(cancellationToken);
var activeSlots = _queue.GetActive()
.Select(a => new QueueSlotDto(a.slot, a.taskId, a.startedAt))
.ToList();
await using var ctx = await _dbFactory.CreateDbContextAsync(cancellationToken);
var now = DateTime.UtcNow;
var waitingTaskIds = await ctx.Tasks
.Where(t => t.Status == TaskStatus.Queued
&& t.BlockedByTaskId == null
&& !t.IsManual
&& (t.ScheduledFor == null || t.ScheduledFor <= now))
.OrderBy(t => t.SortOrder).ThenBy(t => t.CreatedAt)
.Select(t => t.Id)
.ToListAsync(cancellationToken);
return new GetQueueStateResult(configured, effective, activeSlots, waitingTaskIds);
}
}
+11 -6
View File
@@ -12,10 +12,13 @@ public sealed record WaitForTaskChangeResult(IReadOnlyList<TaskStatusChangeDto>
[McpServerToolType] [McpServerToolType]
public sealed class TaskWaitMcpTools public sealed class TaskWaitMcpTools
{ {
// InteractiveLaunchSpecService sets MCP_TOOL_TIMEOUT=200000ms for the list handler // Every ClaudeDo-owned launcher (ClaudeProcess for headless runs, InteractiveLaunchSpecService
// session; this cap leaves a ~30s margin so the tool itself reports TimedOut instead // for ConPTY sessions) sets MCP_TOOL_TIMEOUT=930000ms on the claude CLI process; this cap
// of racing the client's own abort. // leaves a ~30s margin under that so the tool itself reports TimedOut instead of racing the
internal const int MaxTimeoutSeconds = 170; // client's own abort. A caller running claude with a different MCP_TOOL_TIMEOUT (or none --
// the CLI default is 60s) will see its own client-side timeout fire first; this tool has no
// way to detect or compensate for that from the server side.
internal const int MaxTimeoutSeconds = 900;
private static readonly TimeSpan PollInterval = TimeSpan.FromMilliseconds(500); private static readonly TimeSpan PollInterval = TimeSpan.FromMilliseconds(500);
private readonly IDbContextFactory<ClaudeDoDbContext> _dbFactory; private readonly IDbContextFactory<ClaudeDoDbContext> _dbFactory;
@@ -27,9 +30,11 @@ public sealed class TaskWaitMcpTools
[McpServerTool, Description( [McpServerTool, Description(
"Blocks until at least one of the given tasks leaves Queued/Running, or until timeoutSeconds elapses " + "Blocks until at least one of the given tasks leaves Queued/Running, or until timeoutSeconds elapses " +
"(clamped server-side to 170s). Returns immediately if any task is already outside Queued/Running " + "(clamped server-side to 900s). Returns immediately if any task is already outside Queued/Running " +
"when called (an unknown id is reported as status \"NotFound\" and counts as changed). Use this instead " + "when called (an unknown id is reported as status \"NotFound\" and counts as changed). Use this instead " +
"of polling get_task in a loop. Result: { changed: [{ taskId, status }], timedOut }.")] "of polling get_task in a loop. Requires the calling claude process to run with " +
"MCP_TOOL_TIMEOUT >= 930000 (ms) for a long wait to actually be held open -- ClaudeDo's own " +
"launchers already set this. Result: { changed: [{ taskId, status }], timedOut }.")]
public async Task<WaitForTaskChangeResult> WaitForTaskChange( public async Task<WaitForTaskChangeResult> WaitForTaskChange(
string[] taskIds, int timeoutSeconds = 60, CancellationToken cancellationToken = default) string[] taskIds, int timeoutSeconds = 60, CancellationToken cancellationToken = default)
{ {
+2
View File
@@ -302,6 +302,7 @@ if (cfg.ExternalMcpPort > 0)
externalBuilder.Services.AddScoped<LifecycleMcpTools>(); externalBuilder.Services.AddScoped<LifecycleMcpTools>();
externalBuilder.Services.AddScoped<AppSettingsMcpTools>(); externalBuilder.Services.AddScoped<AppSettingsMcpTools>();
externalBuilder.Services.AddScoped<TaskWaitMcpTools>(); externalBuilder.Services.AddScoped<TaskWaitMcpTools>();
externalBuilder.Services.AddScoped<QueueStateMcpTools>();
externalBuilder.Services.AddSingleton(app.Services.GetRequiredService<AttachmentStore>()); externalBuilder.Services.AddSingleton(app.Services.GetRequiredService<AttachmentStore>());
externalBuilder.Services.AddScoped<TaskAttachmentRepository>(); externalBuilder.Services.AddScoped<TaskAttachmentRepository>();
externalBuilder.Services.AddScoped<AttachmentMcpTools>(); externalBuilder.Services.AddScoped<AttachmentMcpTools>();
@@ -317,6 +318,7 @@ if (cfg.ExternalMcpPort > 0)
.WithTools<LifecycleMcpTools>() .WithTools<LifecycleMcpTools>()
.WithTools<AppSettingsMcpTools>() .WithTools<AppSettingsMcpTools>()
.WithTools<TaskWaitMcpTools>() .WithTools<TaskWaitMcpTools>()
.WithTools<QueueStateMcpTools>()
.WithTools<AttachmentMcpTools>(); .WithTools<AttachmentMcpTools>();
externalBuilder.WebHost.UseUrls($"http://127.0.0.1:{cfg.ExternalMcpPort}"); externalBuilder.WebHost.UseUrls($"http://127.0.0.1:{cfg.ExternalMcpPort}");
+10 -8
View File
@@ -125,7 +125,7 @@ public sealed class QueueService : BackgroundService
await Task.WhenAny(wakeTask, timerTask); await Task.WhenAny(wakeTask, timerTask);
var maxParallel = await GetEffectiveMaxParallelAsync(stoppingToken); var (_, maxParallel) = await GetSlotCountsAsync(stoppingToken);
var gateDecision = await _usageGate.EvaluateAsync(stoppingToken); var gateDecision = await _usageGate.EvaluateAsync(stoppingToken);
await ReportUsageGateTransitionAsync(gateDecision); await ReportUsageGateTransitionAsync(gateDecision);
@@ -200,11 +200,13 @@ public sealed class QueueService : BackgroundService
} }
/// <summary> /// <summary>
/// Configured parallelism, stepped down by <see cref="UsageThrottle"/> ahead of the hard usage /// Configured parallelism, and that same value stepped down by <see cref="UsageThrottle"/>
/// gate. A missing snapshot (poll hasn't landed / endpoint unreachable) fails open to the /// ahead of the hard usage gate. A missing snapshot (poll hasn't landed / endpoint
/// configured value — a broken usage poll must never stall the queue. /// unreachable) fails open to the configured value — a broken usage poll must never stall
/// the queue. Also called by <c>get_queue_state</c> (External MCP) to surface the throttle
/// from outside the process.
/// </summary> /// </summary>
private async Task<int> GetEffectiveMaxParallelAsync(CancellationToken ct) public async Task<(int Configured, int Effective)> GetSlotCountsAsync(CancellationToken ct)
{ {
int configured; int configured;
int softPct, hardPct, gateFivePct, gateSevenPct; int softPct, hardPct, gateFivePct, gateSevenPct;
@@ -221,14 +223,14 @@ public sealed class QueueService : BackgroundService
catch (Exception ex) catch (Exception ex)
{ {
_logger.LogWarning(ex, "Failed to read max parallel executions; defaulting to 1"); _logger.LogWarning(ex, "Failed to read max parallel executions; defaulting to 1");
return 1; return (1, 1);
} }
var snapshot = _usageState.Snapshot; var snapshot = _usageState.Snapshot;
if (snapshot is null || _usageState.LastError is not null) if (snapshot is null || _usageState.LastError is not null)
{ {
_lastEffectiveSlots = configured; _lastEffectiveSlots = configured;
return configured; return (configured, configured);
} }
var effective = UsageThrottle.EffectiveSlots( var effective = UsageThrottle.EffectiveSlots(
@@ -236,7 +238,7 @@ public sealed class QueueService : BackgroundService
softPct, hardPct, gateFivePct, gateSevenPct); softPct, hardPct, gateFivePct, gateSevenPct);
ReportThrottleTransition(configured, effective, snapshot); ReportThrottleTransition(configured, effective, snapshot);
return effective; return (configured, effective);
} }
private void ReportThrottleTransition(int configured, int effective, UsageSnapshot snapshot) private void ReportThrottleTransition(int configured, int effective, UsageSnapshot snapshot)
+5 -3
View File
@@ -40,9 +40,11 @@ public sealed class ClaudeProcess : IClaudeProcess
psi.ArgumentList.Add(arg); psi.ArgumentList.Add(arg);
// Claude Code caps HTTP MCP tool calls at 60 s unless MCP_TOOL_TIMEOUT is raised. // Claude Code caps HTTP MCP tool calls at 60 s unless MCP_TOOL_TIMEOUT is raised.
// The in-task AskUser tool blocks up to 3 min waiting for the user, so lift the cap // wait_for_task_change blocks up to TaskWaitMcpTools.MaxTimeoutSeconds (900 s) and the
// (with margin) or that wait would be killed early. Harmless for every other tool. // in-task AskUser tool blocks up to 3 min waiting for the user, so lift the cap well
psi.Environment["MCP_TOOL_TIMEOUT"] = "200000"; // past the longer of the two (with margin) or that wait would be killed early. Harmless
// for every other tool. Keep in sync with InteractiveLaunchSpecService's MCP_TOOL_TIMEOUT.
psi.Environment["MCP_TOOL_TIMEOUT"] = "930000";
using var process = new Process { StartInfo = psi }; using var process = new Process { StartInfo = psi };
process.Start(); process.Start();
@@ -14,7 +14,7 @@ namespace ClaudeDo.Worker.Runner;
// session in an existing task's worktree -- the SAME worktree prep as an autonomous run: // session in an existing task's worktree -- the SAME worktree prep as an autonomous run:
// session-skills seeded onto disk (reuses ISessionSkillSeeder + TaskRunner.UnionSkillNames, // session-skills seeded onto disk (reuses ISessionSkillSeeder + TaskRunner.UnionSkillNames,
// exactly like TaskRunner.RunAsync/ContinueAsync) and the same run environment variables // exactly like TaskRunner.RunAsync/ContinueAsync) and the same run environment variables
// (reuses ClaudeProcess's MCP_TOOL_TIMEOUT). Exe/Args reuse WindowsTerminalLauncher's // (matches ClaudeProcess's MCP_TOOL_TIMEOUT). Exe/Args reuse WindowsTerminalLauncher's
// --resume argument construction. Guards: no running/queued task, and (once a worktree // --resume argument construction. Guards: no running/queued task, and (once a worktree
// exists) it must be live on disk -- but a never-run task (no persisted SessionId) is not // exists) it must be live on disk -- but a never-run task (no persisted SessionId) is not
// an error here, it's a fresh-start spec. // an error here, it's a fresh-start spec.
@@ -107,11 +107,11 @@ public sealed class InteractiveLaunchSpecService : IInteractiveLaunchSpecService
: await BuildFreshTaskArgsAsync(task, effort, ct); : await BuildFreshTaskArgsAsync(task, effort, ct);
// Same run environment variable ClaudeProcess sets for every headless run: the // Same run environment variable ClaudeProcess sets for every headless run: the
// AskUser MCP tool call caps at 60s unless raised, and lifting it is harmless for // AskUser MCP tool call and wait_for_task_change cap at 60s unless raised, and lifting
// every other tool. // it is harmless for every other tool. Keep in sync with ClaudeProcess's MCP_TOOL_TIMEOUT.
var env = new Dictionary<string, string> var env = new Dictionary<string, string>
{ {
["MCP_TOOL_TIMEOUT"] = "200000", ["MCP_TOOL_TIMEOUT"] = "930000",
}; };
return new LaunchSpec(worktree.Path, resolvedClaude, args, env); return new LaunchSpec(worktree.Path, resolvedClaude, args, env);
@@ -129,7 +129,7 @@ public sealed class InteractiveLaunchSpecService : IInteractiveLaunchSpecService
{ {
["MAX_THINKING_TOKENS"] = "20000", ["MAX_THINKING_TOKENS"] = "20000",
["CLAUDEDO_PLANNING_TOKEN"] = ctx.Token, ["CLAUDEDO_PLANNING_TOKEN"] = ctx.Token,
["MCP_TOOL_TIMEOUT"] = "200000", ["MCP_TOOL_TIMEOUT"] = "930000",
}; };
return new LaunchSpec( return new LaunchSpec(
@@ -147,7 +147,7 @@ public sealed class InteractiveLaunchSpecService : IInteractiveLaunchSpecService
var env = new Dictionary<string, string> var env = new Dictionary<string, string>
{ {
["CLAUDEDO_PLANNING_TOKEN"] = ctx.Token, ["CLAUDEDO_PLANNING_TOKEN"] = ctx.Token,
["MCP_TOOL_TIMEOUT"] = "200000", ["MCP_TOOL_TIMEOUT"] = "930000",
}; };
return new LaunchSpec( return new LaunchSpec(
@@ -170,7 +170,7 @@ public sealed class InteractiveLaunchSpecService : IInteractiveLaunchSpecService
var env = new Dictionary<string, string> var env = new Dictionary<string, string>
{ {
["MCP_TOOL_TIMEOUT"] = "200000", ["MCP_TOOL_TIMEOUT"] = "930000",
}; };
// No task and no list here — the global default model's preset decides the effort. // No task and no list here — the global default model's preset decides the effort.
@@ -247,7 +247,7 @@ public sealed class InteractiveLaunchSpecService : IInteractiveLaunchSpecService
var env = new Dictionary<string, string> var env = new Dictionary<string, string>
{ {
["MCP_TOOL_TIMEOUT"] = "200000", ["MCP_TOOL_TIMEOUT"] = "930000",
}; };
return new LaunchSpec(repoDir, resolvedClaude, args, env); return new LaunchSpec(repoDir, resolvedClaude, args, env);
@@ -0,0 +1,173 @@
using ClaudeDo.Data;
using ClaudeDo.Data.Git;
using ClaudeDo.Data.Models;
using ClaudeDo.Data.Repositories;
using ClaudeDo.Worker.Config;
using ClaudeDo.Worker.External;
using ClaudeDo.Worker.Hub;
using ClaudeDo.Worker.Queue;
using ClaudeDo.Worker.Runner;
using ClaudeDo.Worker.Tests.Infrastructure;
using ClaudeDo.Worker.Usage;
using Microsoft.Extensions.Logging.Abstractions;
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
namespace ClaudeDo.Worker.Tests.External;
public sealed class QueueStateMcpToolsTests : IDisposable
{
private readonly DbFixture _db = new();
private readonly ClaudeDoDbContext _ctx;
private readonly TaskRepository _taskRepo;
private readonly ListRepository _listRepo;
private readonly WorkerConfig _cfg;
private readonly string _tempDir;
public QueueStateMcpToolsTests()
{
_ctx = _db.CreateContext();
_taskRepo = new TaskRepository(_ctx);
_listRepo = new ListRepository(_ctx);
_tempDir = Path.Combine(Path.GetTempPath(), $"claudedo_test_{Guid.NewGuid():N}");
Directory.CreateDirectory(_tempDir);
_cfg = new WorkerConfig
{
SandboxRoot = Path.Combine(_tempDir, "sandbox"),
LogRoot = Path.Combine(_tempDir, "logs"),
QueueBackstopIntervalMs = 50,
};
}
public void Dispose()
{
_ctx.Dispose();
_db.Dispose();
try { Directory.Delete(_tempDir, true); } catch { }
}
private (QueueService queue, QueueStateMcpTools sut) CreateSut(
Func<string, string, IReadOnlyList<string>, Func<string, Task>, CancellationToken, Task<RunResult>>? handler = null,
UsageState? usageState = null)
{
var dbFactory = _db.CreateFactory();
var fake = new FakeClaudeProcess(handler);
var broadcaster = new HubBroadcaster(new CapturingHubContext());
var wtManager = new WorktreeManager(new GitService(), dbFactory, _cfg, NullLogger<WorktreeManager>.Instance);
var built = TaskStateServiceBuilder.Build(dbFactory);
var runner = new TaskRunner(fake, dbFactory, broadcaster, wtManager, new ClaudeArgsBuilder(), _cfg,
NullLogger<TaskRunner>.Instance, built.State, new TaskRunTokenRegistry(), new AttachmentStore(),
new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader());
var picker = new QueuePicker(dbFactory);
var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger<OverrideSlotService>.Instance, built.RunCancels);
var queue = new QueueService(dbFactory, runner, _cfg, NullLogger<QueueService>.Instance, new QueueWaker(), picker,
overrideSlot, built.State, built.RunCancels, new FakeUsageGate(), usageState ?? new UsageState(), broadcaster);
return (queue, new QueueStateMcpTools(queue, dbFactory));
}
private async Task<string> SeedListAsync()
{
var listId = Guid.NewGuid().ToString();
await _listRepo.AddAsync(new ListEntity { Id = listId, Name = "Test", CreatedAt = DateTime.UtcNow });
return listId;
}
private async Task<TaskEntity> SeedTaskAsync(
string listId, TaskStatus status, int sortOrder = 0, DateTime? createdAt = null,
bool isManual = false, string? blockedByTaskId = null, DateTime? scheduledFor = null)
{
var task = new TaskEntity
{
Id = Guid.NewGuid().ToString(),
ListId = listId,
Title = "Test task",
Status = status,
SortOrder = sortOrder,
CreatedAt = createdAt ?? DateTime.UtcNow,
IsManual = isManual,
BlockedByTaskId = blockedByTaskId,
ScheduledFor = scheduledFor,
};
// Bypass TaskRepository.AddAsync, which overwrites SortOrder with max(listId)+1 --
// these tests need to control SortOrder directly to exercise queue pick order.
_ctx.Tasks.Add(task);
await _ctx.SaveChangesAsync();
return task;
}
private async Task SetMaxParallelAsync(int maxParallel, int softPct = 0, int hardPct = 0)
{
using var ctx = _db.CreateContext();
var repo = new AppSettingsRepository(ctx);
var settings = await repo.GetAsync();
settings.MaxParallelExecutions = maxParallel;
settings.UsageThrottleSoftPct = softPct;
settings.UsageThrottleHardPct = hardPct;
await repo.UpdateAsync(settings);
}
[Fact]
public async Task GetQueueState_NoThrottle_EffectiveEqualsConfigured()
{
await SetMaxParallelAsync(maxParallel: 3);
var (_, sut) = CreateSut();
var result = await sut.GetQueueState(CancellationToken.None);
Assert.Equal(3, result.ConfiguredSlots);
Assert.Equal(3, result.EffectiveSlots);
}
[Fact]
public async Task GetQueueState_UsageThrottleActive_EffectiveBelowConfigured()
{
await SetMaxParallelAsync(maxParallel: 3, softPct: 50, hardPct: 65);
var usageState = new UsageState();
usageState.ReportSuccess(new UsageSnapshot(
new UsageBucket(60, null), new UsageBucket(0, null), Array.Empty<UsageLimitRow>(), DateTime.UtcNow));
var (_, sut) = CreateSut(usageState: usageState);
var result = await sut.GetQueueState(CancellationToken.None);
Assert.Equal(3, result.ConfiguredSlots);
Assert.Equal(2, result.EffectiveSlots);
}
[Fact]
public async Task GetQueueState_ActiveSlots_ReflectsOverrideSlot()
{
var listId = await SeedListAsync();
var tcs = new TaskCompletionSource<RunResult>();
var (queue, sut) = CreateSut((_, _, _, _, _) => tcs.Task);
var task = await SeedTaskAsync(listId, TaskStatus.Queued);
await queue.RunNow(task.Id);
var result = await sut.GetQueueState(CancellationToken.None);
var slot = Assert.Single(result.ActiveSlots);
Assert.Equal("override", slot.Slot);
Assert.Equal(task.Id, slot.TaskId);
tcs.SetResult(new RunResult { ExitCode = 0, ResultMarkdown = "ok" });
}
[Fact]
public async Task GetQueueState_WaitingTaskIds_OrderedBySortOrderThenCreatedAt_AndFiltersIneligible()
{
var listId = await SeedListAsync();
var (_, sut) = CreateSut();
var second = await SeedTaskAsync(listId, TaskStatus.Queued, sortOrder: 1, createdAt: DateTime.UtcNow);
var first = await SeedTaskAsync(listId, TaskStatus.Queued, sortOrder: 0, createdAt: DateTime.UtcNow.AddMinutes(1));
await SeedTaskAsync(listId, TaskStatus.Queued, sortOrder: 2, isManual: true);
await SeedTaskAsync(listId, TaskStatus.Queued, sortOrder: 3, blockedByTaskId: first.Id);
await SeedTaskAsync(listId, TaskStatus.Queued, sortOrder: 4, scheduledFor: DateTime.UtcNow.AddHours(1));
await SeedTaskAsync(listId, TaskStatus.Running, sortOrder: -1);
var result = await sut.GetQueueState(CancellationToken.None);
Assert.Equal(new[] { first.Id, second.Id }, result.WaitingTaskIds);
}
}
@@ -123,7 +123,7 @@ public sealed class TaskWaitMcpToolsTests : IDisposable
[Fact] [Fact]
public void MaxTimeoutSeconds_StaysComfortablyUnderMcpToolTimeout() public void MaxTimeoutSeconds_StaysComfortablyUnderMcpToolTimeout()
{ {
// InteractiveLaunchSpecService sets MCP_TOOL_TIMEOUT=200000ms for the list handler. // ClaudeProcess / InteractiveLaunchSpecService set MCP_TOOL_TIMEOUT=930000ms.
Assert.True(TaskWaitMcpTools.MaxTimeoutSeconds < 200); Assert.True(TaskWaitMcpTools.MaxTimeoutSeconds < 930);
} }
} }
@@ -241,7 +241,7 @@ public sealed class InteractiveLaunchSpecServiceTests : IDisposable
Assert.Equal(_worktreeDir, spec.Cwd); Assert.Equal(_worktreeDir, spec.Cwd);
Assert.Equal(_claudeStubPath, spec.Exe); Assert.Equal(_claudeStubPath, spec.Exe);
Assert.Equal(new[] { "--resume", "sess-123" }, ArgsAfterEffort(spec)); Assert.Equal(new[] { "--resume", "sess-123" }, ArgsAfterEffort(spec));
Assert.Equal("200000", spec.Env["MCP_TOOL_TIMEOUT"]); Assert.Equal("930000", spec.Env["MCP_TOOL_TIMEOUT"]);
} }
[Fact] [Fact]
@@ -400,7 +400,7 @@ public sealed class InteractiveLaunchSpecServiceTests : IDisposable
Assert.Equal(_tempDir, spec.Cwd); Assert.Equal(_tempDir, spec.Cwd);
Assert.Equal(_claudeStubPath, spec.Exe); Assert.Equal(_claudeStubPath, spec.Exe);
Assert.Empty(ArgsAfterEffort(spec)); Assert.Empty(ArgsAfterEffort(spec));
Assert.Equal("200000", spec.Env["MCP_TOOL_TIMEOUT"]); Assert.Equal("930000", spec.Env["MCP_TOOL_TIMEOUT"]);
Assert.Empty(_seeder.Calls); Assert.Empty(_seeder.Calls);
} }
@@ -534,7 +534,7 @@ public sealed class InteractiveLaunchSpecServiceTests : IDisposable
Assert.Contains(briefPath, kickoff); Assert.Contains(briefPath, kickoff);
Assert.DoesNotContain('\n', kickoff); Assert.DoesNotContain('\n', kickoff);
Assert.Equal("200000", spec.Env["MCP_TOOL_TIMEOUT"]); Assert.Equal("930000", spec.Env["MCP_TOOL_TIMEOUT"]);
} }
[Fact] [Fact]