refactor(interactive): remove streaming interactive stack (superseded by ConPTY)
The embedded ConPTY terminal replaced the in-app streaming interactive session, so delete the dead stack: StreamingClaudeSession, InteractiveSessionService, ProcessClaudeStreamTransport, IClaudeStreamTransport, ILiveSession, LiveSessionRegistry, IdleSessionReaper (+ WorkerConfig.InteractiveIdleTimeoutMinutes), the WorkerHub interactive methods + HubBroadcaster events, IWorkerClient interactive members, the TaskMonitorViewModel composer + SessionTerminalView composer markup, and the old 'Run interactively' entry. AskUser/PendingQuestionRegistry, the autonomous path, planning, ResumeTaskInTerminal, and all ConPTY code are kept. Localization pruned.
This commit is contained in:
mika kuns
@@ -12,7 +12,7 @@ Worker/
|
||||
Worktrees/ — WorktreeMaintenanceService
|
||||
Agents/ — AgentFileService, DefaultAgentSeeder
|
||||
Runner/ — TaskRunner + Claude CLI integration; TaskRunMcpService/TaskRunMcpContext/TaskRunTokenRegistry (in-task MCP wired during execution)
|
||||
Planning/ — PlanningSessionManager, PlanningChainCoordinator, PlanningMcpService, PlanningMergeOrchestrator, PlanningAggregator, PlanningSessionContext/PlanningTokenAuth/PlanningMcpContextAccessor, WindowsTerminalLauncher (ITerminalLauncher) — wt launcher for planning + interactive sessions
|
||||
Planning/ — PlanningSessionManager, PlanningChainCoordinator, PlanningMcpService, PlanningMergeOrchestrator, PlanningAggregator, PlanningSessionContext/PlanningTokenAuth/PlanningMcpContextAccessor, WindowsTerminalLauncher (ITerminalLauncher) — wt launcher for planning sessions + pick-up-in-terminal
|
||||
Refine/ — RefineRunner + RefinePrompt (hub `RefineTask`; broadcasts RefineStarted/RefineFinished)
|
||||
External/ — ExternalMcpService + sibling tool classes
|
||||
Config/ — WorkerConfig
|
||||
@@ -154,7 +154,8 @@ Each CLI invocation is recorded in the `task_runs` table via `TaskRunRepository`
|
||||
- Execution: `Ping`, `GetActive`, `RunNow`, `CancelTask`, `WakeQueue`, `ContinueTask`, `ResetTask`, `SetTaskStatus`, `RefineTask`
|
||||
- Review/merge: `ApproveReview(taskId, targetBranch) -> MergeResultDto` (childless task: merges its worktree then Done, conflict stays WaitingForReview; task with children: drives `PlanningMergeOrchestrator` to merge the whole unit), `ContinuePlanningMerge` / `AbortPlanningMerge` (resolve a unit-merge conflict), `PreviewMerge(taskId, targetBranch) -> MergePreviewDto` (non-destructive mergeability check), `RejectReviewToQueue`, `RejectReviewToIdle`, `CancelReview`, `MergeTask`, `GetMergeTargets`
|
||||
- Single-task conflict resolver (Layer C): `StartConflictMerge`, `GetMergeConflictDocuments` (segments), `WriteConflictResolution`, `ContinueConflictMerge`, `AbortConflictMerge` (service-level `TaskMergeService.ContinueMergeAsync`/`AbortMergeAsync` keep their names)
|
||||
- Planning sessions: `StartPlanningSession`, `ResumePlanningSession`, `DiscardPlanningSession`, `FinalizePlanningSession`, `QueuePlanningSubtasks`, `GetPendingDraftCount`, `OpenInteractiveTerminal`, `GetPlanningAggregate` (per-subtask diffs), `BuildPlanningIntegrationBranch` (combined diff)
|
||||
- Planning sessions: `StartPlanningSession`, `ResumePlanningSession`, `DiscardPlanningSession`, `FinalizePlanningSession`, `QueuePlanningSubtasks`, `GetPendingDraftCount`, `GetPlanningAggregate` (per-subtask diffs), `BuildPlanningIntegrationBranch` (combined diff)
|
||||
- Interactive sessions (embedded ConPTY, UI process): `ResumeTaskInTerminal` (pick-up-in-terminal), `GetInteractiveLaunchSpec`, `GetAdHocLaunchSpec`
|
||||
- Worktrees: `CleanupFinishedWorktrees`, `ResetAllWorktrees`, `GetWorktreesOverview`, `SetWorktreeState`, `ForceRemoveWorktree`
|
||||
- Agents/settings/lists: `GetAgents`, `RefreshAgents`, `RestoreDefaultAgents`, `GetAppSettings`, `UpdateAppSettings`, `UpdateList`, `UpdateListConfig`, `GetListConfig`, `UpdateTaskAgentSettings`
|
||||
- Reports/notes/prep: `GetWeekReport`, `GenerateWeekReport`, `GetDailyNotes`, `AddDailyNote`, `UpdateDailyNote`, `DeleteDailyNote`, `RunDailyPrepNow`, `ClearMyDay`, `GetLastPrepLog`, `ListPrimeSchedules`, `UpsertPrimeSchedule`, `DeletePrimeSchedule`
|
||||
|
||||
@@ -41,10 +41,6 @@ public sealed class WorkerConfig
|
||||
[JsonPropertyName("external_mcp_api_key")]
|
||||
public string? ExternalMcpApiKey { get; set; }
|
||||
|
||||
/// <summary>Interactive/streaming sessions idle longer than this are stopped by IdleSessionReaper. 0 disables reaping.</summary>
|
||||
[JsonPropertyName("interactive_idle_timeout_minutes")]
|
||||
public int InteractiveIdleTimeoutMinutes { get; set; } = 30;
|
||||
|
||||
[JsonPropertyName("online_inbox")]
|
||||
public OnlineInboxConfig OnlineInbox { get; set; } = new();
|
||||
|
||||
|
||||
@@ -77,16 +77,4 @@ public sealed class HubBroadcaster : IPrimeBroadcaster, IRefineBroadcaster
|
||||
Task IRefineBroadcaster.RefineStartedAsync(string taskId) => RefineStarted(taskId);
|
||||
Task IRefineBroadcaster.RefineFinishedAsync(string taskId, bool success, string? error) =>
|
||||
RefineFinished(taskId, success, error);
|
||||
|
||||
public Task InteractiveSessionStarted(string taskId) =>
|
||||
_hub.Clients.All.SendAsync("InteractiveSessionStarted", taskId);
|
||||
|
||||
public Task InteractiveSessionEnded(string taskId) =>
|
||||
_hub.Clients.All.SendAsync("InteractiveSessionEnded", taskId);
|
||||
|
||||
public Task InteractiveQueueChanged(string taskId, IReadOnlyList<string> pending) =>
|
||||
_hub.Clients.All.SendAsync("InteractiveQueueChanged", taskId, pending);
|
||||
|
||||
public Task InteractiveMessageSent(string taskId, string text) =>
|
||||
_hub.Clients.All.SendAsync("InteractiveMessageSent", taskId, text);
|
||||
}
|
||||
|
||||
@@ -127,7 +127,6 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
private readonly OnlineInboxConfig _onlineInboxConfig;
|
||||
private readonly OnlineTokenStore _onlineTokenStore;
|
||||
private readonly Runner.PendingQuestionRegistry _pendingQuestions;
|
||||
private readonly InteractiveSessionService _interactive;
|
||||
private readonly LogRingBuffer? _logBuffer;
|
||||
private readonly ISessionSkillRegistry _skillRegistry;
|
||||
private readonly IInteractiveLaunchSpecService? _interactiveLaunchSpec;
|
||||
@@ -156,7 +155,6 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
OnlineInboxConfig onlineInboxConfig,
|
||||
OnlineTokenStore onlineTokenStore,
|
||||
Runner.PendingQuestionRegistry pendingQuestions,
|
||||
InteractiveSessionService interactive,
|
||||
ISessionSkillRegistry skillRegistry,
|
||||
LogRingBuffer? logBuffer = null,
|
||||
IInteractiveLaunchSpecService? interactiveLaunchSpec = null)
|
||||
@@ -184,7 +182,6 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
_onlineInboxConfig = onlineInboxConfig;
|
||||
_onlineTokenStore = onlineTokenStore;
|
||||
_pendingQuestions = pendingQuestions;
|
||||
_interactive = interactive;
|
||||
_skillRegistry = skillRegistry;
|
||||
_logBuffer = logBuffer;
|
||||
_interactiveLaunchSpec = interactiveLaunchSpec;
|
||||
@@ -624,13 +621,9 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
return ctx;
|
||||
}
|
||||
|
||||
public Task OpenInteractiveTerminalAsync(string taskId) =>
|
||||
_interactive.StartAsync(taskId, Context.ConnectionAborted);
|
||||
|
||||
// Picks up a task's Claude session in a real terminal window (--resume) so the user can
|
||||
// drive it by hand — distinct from OpenInteractiveTerminalAsync (the in-app streaming
|
||||
// session). Only for tasks the worker isn't actively running, with a persisted session
|
||||
// id and a live worktree.
|
||||
// drive it by hand. Only for tasks the worker isn't actively running, with a persisted
|
||||
// session id and a live worktree.
|
||||
public Task ResumeTaskInTerminal(string taskId) => HubGuard(async () =>
|
||||
{
|
||||
await using var ctx = await _dbFactory.CreateDbContextAsync();
|
||||
@@ -680,18 +673,6 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
return _interactiveLaunchSpec.BuildForDirectoryAsync(directory, Context.ConnectionAborted);
|
||||
});
|
||||
|
||||
public Task SendInteractiveMessage(string taskId, string text) =>
|
||||
_interactive.SendAsync(taskId, text, Context.ConnectionAborted);
|
||||
|
||||
public Task StopInteractiveSession(string taskId) =>
|
||||
_interactive.StopAsync(taskId, Context.ConnectionAborted);
|
||||
|
||||
public Task InterruptInteractiveSession(string taskId) =>
|
||||
_interactive.InterruptAsync(taskId, Context.ConnectionAborted);
|
||||
|
||||
public Task RemoveQueuedInteractiveMessage(string taskId, string text) =>
|
||||
_interactive.RemoveQueuedAsync(taskId, text, Context.ConnectionAborted);
|
||||
|
||||
public async Task<DiscardPlanningOutcome> DiscardPlanningSessionAsync(string taskId, bool dequeueQueuedChildren = false)
|
||||
{
|
||||
var outcome = await _planning.DiscardAsync(taskId, dequeueQueuedChildren, Context.ConnectionAborted);
|
||||
|
||||
@@ -1,172 +0,0 @@
|
||||
using System.Text;
|
||||
using ClaudeDo.Data;
|
||||
using ClaudeDo.Data.Models;
|
||||
using ClaudeDo.Data.Repositories;
|
||||
using ClaudeDo.Worker.Config;
|
||||
using ClaudeDo.Worker.Hub;
|
||||
using ClaudeDo.Worker.Runner;
|
||||
using ClaudeDo.Worker.Runner.Interfaces;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
|
||||
namespace ClaudeDo.Worker.Planning;
|
||||
|
||||
public sealed class InteractiveSessionService
|
||||
{
|
||||
private readonly IDbContextFactory<ClaudeDoDbContext> _dbFactory;
|
||||
private readonly WorkerConfig _cfg;
|
||||
private readonly HubBroadcaster _broadcaster;
|
||||
private readonly LiveSessionRegistry _registry;
|
||||
private readonly ILoggerFactory _loggerFactory;
|
||||
|
||||
// Optional factory for tests. Signature: (onLine) -> (session, waitForExitTask).
|
||||
// The waitForExitTask completes when the underlying process has exited.
|
||||
private readonly Func<string, IReadOnlyList<string>, Func<string, Task>, (ILiveSession session, Task exitTask)>? _sessionFactory;
|
||||
|
||||
public InteractiveSessionService(
|
||||
IDbContextFactory<ClaudeDoDbContext> dbFactory,
|
||||
WorkerConfig cfg,
|
||||
HubBroadcaster broadcaster,
|
||||
LiveSessionRegistry registry,
|
||||
ILoggerFactory loggerFactory,
|
||||
Func<string, IReadOnlyList<string>, Func<string, Task>, (ILiveSession session, Task exitTask)>? sessionFactory = null)
|
||||
{
|
||||
_dbFactory = dbFactory;
|
||||
_cfg = cfg;
|
||||
_broadcaster = broadcaster;
|
||||
_registry = registry;
|
||||
_loggerFactory = loggerFactory;
|
||||
_sessionFactory = sessionFactory;
|
||||
}
|
||||
|
||||
public async Task StartAsync(string taskId, CancellationToken ct)
|
||||
{
|
||||
if (_registry.TryGet(taskId, out _))
|
||||
throw new InvalidOperationException("An interactive session is already running for this task.");
|
||||
|
||||
await using var ctx = _dbFactory.CreateDbContext();
|
||||
var tasks = new TaskRepository(ctx);
|
||||
var lists = new ListRepository(ctx);
|
||||
|
||||
var task = await tasks.GetByIdAsync(taskId, ct)
|
||||
?? throw new InvalidOperationException($"Task {taskId} not found.");
|
||||
var list = await lists.GetByIdAsync(task.ListId, ct)
|
||||
?? throw new InvalidOperationException($"List {task.ListId} not found.");
|
||||
|
||||
var workingDir = list.WorkingDir;
|
||||
if (string.IsNullOrWhiteSpace(workingDir) || !Directory.Exists(workingDir))
|
||||
throw new InvalidOperationException(
|
||||
$"List '{list.Name}' has no valid working directory configured.");
|
||||
|
||||
var seededPrompt = BuildInteractivePrompt(task);
|
||||
|
||||
var args = new[]
|
||||
{
|
||||
"-p",
|
||||
"--input-format", "stream-json",
|
||||
"--output-format", "stream-json",
|
||||
"--verbose",
|
||||
"--replay-user-messages",
|
||||
"--model", ModelRegistry.PlanningAlias,
|
||||
"--permission-mode", "auto",
|
||||
};
|
||||
|
||||
Func<string, Task> onLine = line =>
|
||||
{
|
||||
_registry.Touch(taskId);
|
||||
return _broadcaster.TaskMessage(taskId, "[stdout] " + line);
|
||||
};
|
||||
|
||||
ILiveSession session;
|
||||
Task exitTask;
|
||||
|
||||
if (_sessionFactory is not null)
|
||||
{
|
||||
// Factory is responsible for providing a ready-to-use session and its exit signal.
|
||||
(session, exitTask) = _sessionFactory(workingDir, args, onLine);
|
||||
}
|
||||
else
|
||||
{
|
||||
var transport = new ProcessClaudeStreamTransport(
|
||||
_cfg,
|
||||
_loggerFactory.CreateLogger<ProcessClaudeStreamTransport>());
|
||||
var streamingSession = new StreamingClaudeSession(
|
||||
transport,
|
||||
onLine,
|
||||
_loggerFactory.CreateLogger<StreamingClaudeSession>(),
|
||||
onQueueChanged: pending => _ = _broadcaster.InteractiveQueueChanged(taskId, pending),
|
||||
onUserMessageSent: text => _ = _broadcaster.InteractiveMessageSent(taskId, text));
|
||||
await streamingSession.StartAsync(args, workingDir, seededPrompt, ct);
|
||||
session = streamingSession;
|
||||
exitTask = transport.WaitForExitAsync();
|
||||
}
|
||||
_registry.Register(taskId, session);
|
||||
await _broadcaster.InteractiveSessionStarted(taskId);
|
||||
|
||||
var logger = _loggerFactory.CreateLogger<InteractiveSessionService>();
|
||||
_ = WatchExitAsync(taskId, exitTask, logger);
|
||||
}
|
||||
|
||||
private async Task WatchExitAsync(string taskId, Task exitTask, ILogger logger)
|
||||
{
|
||||
try
|
||||
{
|
||||
await exitTask;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogWarning(ex, "Interactive session exit watcher caught an exception for task {task_id}", taskId);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_registry.Unregister(taskId);
|
||||
try { await _broadcaster.InteractiveSessionEnded(taskId); }
|
||||
catch (Exception ex) { logger.LogWarning(ex, "InteractiveSessionEnded broadcast failed for task {task_id}", taskId); }
|
||||
}
|
||||
}
|
||||
|
||||
public async Task SendAsync(string taskId, string text, CancellationToken ct)
|
||||
{
|
||||
if (!_registry.TryGet(taskId, out var session))
|
||||
throw new InvalidOperationException("No interactive session is running for this task.");
|
||||
_registry.Touch(taskId);
|
||||
await session.SendUserMessageAsync(text, ct);
|
||||
}
|
||||
|
||||
public async Task RemoveQueuedAsync(string taskId, string text, CancellationToken ct)
|
||||
{
|
||||
if (_registry.TryGet(taskId, out var session))
|
||||
{
|
||||
_registry.Touch(taskId);
|
||||
await session.RemoveQueuedAsync(text, ct);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task InterruptAsync(string taskId, CancellationToken ct)
|
||||
{
|
||||
if (_registry.TryGet(taskId, out var session))
|
||||
{
|
||||
_registry.Touch(taskId);
|
||||
await session.InterruptAsync(ct);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task StopAsync(string taskId, CancellationToken ct)
|
||||
{
|
||||
// StopAsync removes from registry and kills the session.
|
||||
// The exit watcher will fire InteractiveSessionEnded once the process exits,
|
||||
// so we don't broadcast here — the watcher is the single authoritative source.
|
||||
await _registry.StopAsync(taskId);
|
||||
}
|
||||
|
||||
private static string BuildInteractivePrompt(TaskEntity task)
|
||||
{
|
||||
var sb = new StringBuilder();
|
||||
sb.AppendLine($"# Task: {task.Title}");
|
||||
if (!string.IsNullOrWhiteSpace(task.Description))
|
||||
{
|
||||
sb.AppendLine();
|
||||
sb.AppendLine(task.Description);
|
||||
}
|
||||
return sb.ToString();
|
||||
}
|
||||
}
|
||||
@@ -2,7 +2,7 @@ namespace ClaudeDo.Worker.Planning;
|
||||
|
||||
// Launches the Claude CLI in a visible terminal for human-driven planning sessions.
|
||||
// Not used for headless task execution (that path is ClaudeProcess, prompt over stdin)
|
||||
// nor for interactive sessions (those use InteractiveSessionService + StreamingClaudeSession).
|
||||
// nor for embedded ConPTY interactive sessions (those use IInteractiveLaunchSpecService).
|
||||
public interface ITerminalLauncher
|
||||
{
|
||||
Task LaunchPlanningStartAsync(PlanningSessionStartContext ctx, CancellationToken cancellationToken);
|
||||
|
||||
@@ -85,9 +85,6 @@ builder.Services.AddSingleton<TaskMergeService>();
|
||||
builder.Services.AddSingleton<PlanningAggregator>();
|
||||
builder.Services.AddSingleton<PlanningMergeOrchestrator>();
|
||||
builder.Services.AddSingleton<PlanningChainCoordinator>();
|
||||
builder.Services.AddSingleton<LiveSessionRegistry>();
|
||||
builder.Services.AddSingleton<InteractiveSessionService>();
|
||||
builder.Services.AddHostedService<IdleSessionReaper>();
|
||||
|
||||
// Queue dispatch primitives. QueueWaker holds the wake semaphore; the queue picker
|
||||
// performs atomic Queued→Running claim. Both injected into the state service so it
|
||||
|
||||
@@ -1,49 +0,0 @@
|
||||
using ClaudeDo.Worker.Config;
|
||||
|
||||
namespace ClaudeDo.Worker.Runner;
|
||||
|
||||
// Stops interactive/streaming sessions that have gone idle. Interactive `claude` processes wait
|
||||
// on stdin and never exit on their own, and there is no client-disconnect teardown — so an
|
||||
// abandoned chat (UI closed, navigated away, crashed) keeps its claude.exe (+ conhost) alive for
|
||||
// the worker's entire lifetime. Under a long-running autostart worker these pile up (observed:
|
||||
// ~170 child processes). This sweep reaps the idle ones.
|
||||
public sealed class IdleSessionReaper : BackgroundService
|
||||
{
|
||||
private static readonly TimeSpan SweepInterval = TimeSpan.FromMinutes(5);
|
||||
|
||||
private readonly LiveSessionRegistry _registry;
|
||||
private readonly WorkerConfig _cfg;
|
||||
private readonly ILogger<IdleSessionReaper> _logger;
|
||||
|
||||
public IdleSessionReaper(LiveSessionRegistry registry, WorkerConfig cfg, ILogger<IdleSessionReaper> logger)
|
||||
{
|
||||
_registry = registry;
|
||||
_cfg = cfg;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
var idleTimeout = TimeSpan.FromMinutes(_cfg.InteractiveIdleTimeoutMinutes);
|
||||
if (idleTimeout <= TimeSpan.Zero)
|
||||
return; // reaper disabled
|
||||
|
||||
using var timer = new PeriodicTimer(SweepInterval);
|
||||
while (await timer.WaitForNextTickAsync(stoppingToken))
|
||||
{
|
||||
try
|
||||
{
|
||||
var reaped = await _registry.ReapIdleAsync(DateTime.UtcNow, idleTimeout);
|
||||
if (reaped.Count > 0)
|
||||
_logger.LogInformation(
|
||||
"Reaped {session_count} idle interactive session(s) after {idle_minutes} min: {task_ids}",
|
||||
reaped.Count, _cfg.InteractiveIdleTimeoutMinutes, string.Join(", ", reaped));
|
||||
}
|
||||
catch (OperationCanceledException) { throw; }
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Idle session reap sweep failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,11 +0,0 @@
|
||||
namespace ClaudeDo.Worker.Runner.Interfaces;
|
||||
|
||||
public interface IClaudeStreamTransport : IAsyncDisposable
|
||||
{
|
||||
Task StartAsync(IReadOnlyList<string> args, string workingDirectory, CancellationToken ct);
|
||||
Task WriteLineAsync(string jsonLine, CancellationToken ct);
|
||||
event Func<string, Task>? LineReceived;
|
||||
event Func<string, Task>? StderrReceived;
|
||||
void Kill();
|
||||
Task WaitForExitAsync();
|
||||
}
|
||||
@@ -1,10 +0,0 @@
|
||||
namespace ClaudeDo.Worker.Runner.Interfaces;
|
||||
|
||||
public interface ILiveSession : IAsyncDisposable
|
||||
{
|
||||
bool IsTurnInFlight { get; }
|
||||
Task SendUserMessageAsync(string text, CancellationToken ct);
|
||||
Task RemoveQueuedAsync(string text, CancellationToken ct);
|
||||
Task InterruptAsync(CancellationToken ct);
|
||||
Task StopAsync();
|
||||
}
|
||||
@@ -1,87 +0,0 @@
|
||||
using System.Collections.Concurrent;
|
||||
using ClaudeDo.Worker.Runner.Interfaces;
|
||||
|
||||
namespace ClaudeDo.Worker.Runner;
|
||||
|
||||
// Singleton in-memory registry of active live streaming sessions.
|
||||
// A session's lifetime matches its associated task run; dead entries are removed by the runner.
|
||||
//
|
||||
// Interactive (stream-json) sessions never exit on their own — they wait on stdin — and there is
|
||||
// no client-disconnect teardown, so an abandoned chat would otherwise keep its claude.exe alive
|
||||
// for the worker's whole lifetime. IdleSessionReaper periodically stops sessions that have seen
|
||||
// no activity past a timeout (see ReapIdleAsync); Touch() records that activity.
|
||||
public sealed class LiveSessionRegistry
|
||||
{
|
||||
private sealed class Entry
|
||||
{
|
||||
public required ILiveSession Session { get; init; }
|
||||
public long LastActivityTicksUtc;
|
||||
}
|
||||
|
||||
private readonly ConcurrentDictionary<string, Entry> _sessions = new();
|
||||
|
||||
public void Register(string taskId, ILiveSession session)
|
||||
{
|
||||
if (_sessions.TryRemove(taskId, out var existing))
|
||||
{
|
||||
// Best-effort stop of the replaced session; don't await to avoid deadlock risk.
|
||||
_ = existing.Session.StopAsync().ContinueWith(t =>
|
||||
{
|
||||
if (t.IsFaulted) { /* swallow — old session is already orphaned */ }
|
||||
}, TaskScheduler.Default);
|
||||
}
|
||||
_sessions[taskId] = new Entry { Session = session, LastActivityTicksUtc = DateTime.UtcNow.Ticks };
|
||||
}
|
||||
|
||||
// Marks a session as active so the idle reaper leaves it alone. Called on every user
|
||||
// message and every output line. No-op if the session is not (yet) registered.
|
||||
public void Touch(string taskId)
|
||||
{
|
||||
if (_sessions.TryGetValue(taskId, out var entry))
|
||||
Interlocked.Exchange(ref entry.LastActivityTicksUtc, DateTime.UtcNow.Ticks);
|
||||
}
|
||||
|
||||
public bool TryGet(string taskId, out ILiveSession session)
|
||||
{
|
||||
if (_sessions.TryGetValue(taskId, out var entry))
|
||||
{
|
||||
session = entry.Session;
|
||||
return true;
|
||||
}
|
||||
session = null!;
|
||||
return false;
|
||||
}
|
||||
|
||||
public void Unregister(string taskId) => _sessions.TryRemove(taskId, out _);
|
||||
|
||||
public async Task StopAsync(string taskId)
|
||||
{
|
||||
if (_sessions.TryRemove(taskId, out var entry))
|
||||
await entry.Session.StopAsync();
|
||||
}
|
||||
|
||||
// Stops and removes every session whose last activity is older than (nowUtc - idleTimeout),
|
||||
// skipping any session with a turn in flight (an agent that's actively working, even if quiet).
|
||||
// Returns the reaped task ids.
|
||||
public async Task<IReadOnlyList<string>> ReapIdleAsync(DateTime nowUtc, TimeSpan idleTimeout)
|
||||
{
|
||||
var cutoffTicks = (nowUtc - idleTimeout).Ticks;
|
||||
List<string>? reaped = null;
|
||||
|
||||
foreach (var kvp in _sessions)
|
||||
{
|
||||
var entry = kvp.Value;
|
||||
if (entry.Session.IsTurnInFlight) continue;
|
||||
if (Interlocked.Read(ref entry.LastActivityTicksUtc) > cutoffTicks) continue;
|
||||
|
||||
if (_sessions.TryRemove(kvp.Key, out var removed))
|
||||
{
|
||||
try { await removed.Session.StopAsync(); }
|
||||
catch { /* already dead — leave it removed */ }
|
||||
(reaped ??= new()).Add(kvp.Key);
|
||||
}
|
||||
}
|
||||
|
||||
return reaped ?? (IReadOnlyList<string>)Array.Empty<string>();
|
||||
}
|
||||
}
|
||||
@@ -1,111 +0,0 @@
|
||||
using System.Diagnostics;
|
||||
using System.Text;
|
||||
using ClaudeDo.Worker.Config;
|
||||
using ClaudeDo.Worker.Runner.Interfaces;
|
||||
|
||||
namespace ClaudeDo.Worker.Runner;
|
||||
|
||||
public sealed class ProcessClaudeStreamTransport : IClaudeStreamTransport
|
||||
{
|
||||
private readonly WorkerConfig _cfg;
|
||||
private readonly ILogger<ProcessClaudeStreamTransport> _logger;
|
||||
|
||||
private Process? _process;
|
||||
private Task? _stdoutTask;
|
||||
private Task? _stderrTask;
|
||||
|
||||
public event Func<string, Task>? LineReceived;
|
||||
public event Func<string, Task>? StderrReceived;
|
||||
|
||||
public ProcessClaudeStreamTransport(WorkerConfig cfg, ILogger<ProcessClaudeStreamTransport> logger)
|
||||
{
|
||||
_cfg = cfg;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public Task StartAsync(IReadOnlyList<string> args, string workingDirectory, CancellationToken ct)
|
||||
{
|
||||
var psi = new ProcessStartInfo
|
||||
{
|
||||
FileName = _cfg.ClaudeBin,
|
||||
WorkingDirectory = workingDirectory,
|
||||
RedirectStandardInput = true,
|
||||
RedirectStandardOutput = true,
|
||||
RedirectStandardError = true,
|
||||
UseShellExecute = false,
|
||||
CreateNoWindow = true,
|
||||
StandardOutputEncoding = Encoding.UTF8,
|
||||
StandardErrorEncoding = Encoding.UTF8,
|
||||
};
|
||||
|
||||
foreach (var arg in args)
|
||||
psi.ArgumentList.Add(arg);
|
||||
|
||||
psi.Environment["MCP_TOOL_TIMEOUT"] = "200000";
|
||||
|
||||
_process = new Process { StartInfo = psi };
|
||||
_process.Start();
|
||||
ProcessJobObject.Assign(_process, _logger);
|
||||
|
||||
// Keep stdin open — turns are driven by WriteLineAsync calls.
|
||||
_process.StandardInput.AutoFlush = false;
|
||||
|
||||
_stdoutTask = Task.Run(async () =>
|
||||
{
|
||||
while (await _process.StandardOutput.ReadLineAsync() is { } line)
|
||||
{
|
||||
if (string.IsNullOrEmpty(line)) continue;
|
||||
var handler = LineReceived;
|
||||
if (handler is not null)
|
||||
{
|
||||
try { await handler(line); }
|
||||
catch (Exception ex) { _logger.LogWarning(ex, "LineReceived handler threw"); }
|
||||
}
|
||||
}
|
||||
}, CancellationToken.None);
|
||||
|
||||
_stderrTask = Task.Run(async () =>
|
||||
{
|
||||
while (await _process.StandardError.ReadLineAsync() is { } line)
|
||||
{
|
||||
if (string.IsNullOrEmpty(line)) continue;
|
||||
var handler = StderrReceived;
|
||||
if (handler is not null)
|
||||
{
|
||||
try { await handler(line); }
|
||||
catch (Exception ex) { _logger.LogWarning(ex, "StderrReceived handler threw"); }
|
||||
}
|
||||
}
|
||||
}, CancellationToken.None);
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task WriteLineAsync(string jsonLine, CancellationToken ct)
|
||||
{
|
||||
if (_process is null) throw new InvalidOperationException("Transport not started.");
|
||||
await _process.StandardInput.WriteAsync((jsonLine + "\n").AsMemory(), ct);
|
||||
await _process.StandardInput.FlushAsync(ct);
|
||||
}
|
||||
|
||||
public void Kill()
|
||||
{
|
||||
try { _process?.Kill(entireProcessTree: true); }
|
||||
catch { /* already exited */ }
|
||||
}
|
||||
|
||||
public async Task WaitForExitAsync()
|
||||
{
|
||||
if (_process is not null)
|
||||
await _process.WaitForExitAsync(CancellationToken.None);
|
||||
if (_stdoutTask is not null) await _stdoutTask;
|
||||
if (_stderrTask is not null) await _stderrTask;
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
Kill();
|
||||
await WaitForExitAsync();
|
||||
_process?.Dispose();
|
||||
}
|
||||
}
|
||||
@@ -1,207 +0,0 @@
|
||||
using System.Text.Json;
|
||||
using ClaudeDo.Worker.Runner.Interfaces;
|
||||
|
||||
namespace ClaudeDo.Worker.Runner;
|
||||
|
||||
public sealed class StreamingClaudeSession : ILiveSession
|
||||
{
|
||||
private readonly IClaudeStreamTransport _transport;
|
||||
private readonly Func<string, Task> _onLine;
|
||||
private readonly ILogger<StreamingClaudeSession> _logger;
|
||||
private readonly Action<IReadOnlyList<string>>? _onQueueChanged;
|
||||
private readonly Action<string>? _onUserMessageSent;
|
||||
|
||||
private readonly SemaphoreSlim _sendLock = new(1, 1);
|
||||
private volatile bool _isTurnInFlight;
|
||||
private readonly Queue<string> _pending = new();
|
||||
|
||||
public bool IsTurnInFlight => _isTurnInFlight;
|
||||
|
||||
public StreamingClaudeSession(
|
||||
IClaudeStreamTransport transport,
|
||||
Func<string, Task> onLine,
|
||||
ILogger<StreamingClaudeSession> logger,
|
||||
Action<IReadOnlyList<string>>? onQueueChanged = null,
|
||||
Action<string>? onUserMessageSent = null)
|
||||
{
|
||||
_transport = transport;
|
||||
_onLine = onLine;
|
||||
_logger = logger;
|
||||
_onQueueChanged = onQueueChanged;
|
||||
_onUserMessageSent = onUserMessageSent;
|
||||
}
|
||||
|
||||
private IReadOnlyList<string> SnapshotPending() => _pending.ToArray();
|
||||
|
||||
public async Task StartAsync(
|
||||
IReadOnlyList<string> args,
|
||||
string workingDirectory,
|
||||
string firstPrompt,
|
||||
CancellationToken ct)
|
||||
{
|
||||
_transport.LineReceived += HandleLineAsync;
|
||||
await _transport.StartAsync(args, workingDirectory, ct);
|
||||
await SendTurnAsync(firstPrompt, ct);
|
||||
_onUserMessageSent?.Invoke(firstPrompt);
|
||||
}
|
||||
|
||||
private async Task HandleLineAsync(string line)
|
||||
{
|
||||
try { await _onLine(line); }
|
||||
catch (Exception ex) { _logger.LogWarning(ex, "onLine callback threw"); }
|
||||
|
||||
bool isResult;
|
||||
try
|
||||
{
|
||||
using var doc = JsonDocument.Parse(line);
|
||||
isResult = doc.RootElement.TryGetProperty("type", out var typeProp)
|
||||
&& typeProp.GetString() == "result";
|
||||
}
|
||||
catch { isResult = false; }
|
||||
|
||||
if (!isResult) return;
|
||||
|
||||
// Turn ended — flush one queued message if available.
|
||||
string? flushedText = null;
|
||||
IReadOnlyList<string>? remainingSnapshot = null;
|
||||
|
||||
await _sendLock.WaitAsync();
|
||||
try
|
||||
{
|
||||
_isTurnInFlight = false;
|
||||
if (_pending.Count > 0)
|
||||
{
|
||||
flushedText = _pending.Dequeue();
|
||||
remainingSnapshot = SnapshotPending();
|
||||
await SendTurnAsync(flushedText, CancellationToken.None);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
_sendLock.Release();
|
||||
}
|
||||
|
||||
if (flushedText is not null)
|
||||
{
|
||||
_onQueueChanged?.Invoke(remainingSnapshot!);
|
||||
_onUserMessageSent?.Invoke(flushedText);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task SendUserMessageAsync(string text, CancellationToken ct)
|
||||
{
|
||||
bool enqueued = false;
|
||||
IReadOnlyList<string>? snapshot = null;
|
||||
|
||||
await _sendLock.WaitAsync(ct);
|
||||
try
|
||||
{
|
||||
if (_isTurnInFlight || _pending.Count > 0)
|
||||
{
|
||||
_pending.Enqueue(text);
|
||||
snapshot = SnapshotPending();
|
||||
enqueued = true;
|
||||
}
|
||||
else
|
||||
{
|
||||
await SendTurnAsync(text, ct);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
_sendLock.Release();
|
||||
}
|
||||
|
||||
if (enqueued)
|
||||
_onQueueChanged?.Invoke(snapshot!);
|
||||
else
|
||||
_onUserMessageSent?.Invoke(text);
|
||||
}
|
||||
|
||||
public async Task RemoveQueuedAsync(string text, CancellationToken ct)
|
||||
{
|
||||
IReadOnlyList<string>? snapshot = null;
|
||||
|
||||
await _sendLock.WaitAsync(ct);
|
||||
try
|
||||
{
|
||||
if (_pending.Count == 0) return;
|
||||
|
||||
var list = _pending.ToList();
|
||||
var idx = list.IndexOf(text);
|
||||
if (idx < 0) return;
|
||||
|
||||
list.RemoveAt(idx);
|
||||
_pending.Clear();
|
||||
foreach (var item in list)
|
||||
_pending.Enqueue(item);
|
||||
|
||||
snapshot = SnapshotPending();
|
||||
}
|
||||
finally
|
||||
{
|
||||
_sendLock.Release();
|
||||
}
|
||||
|
||||
if (snapshot is not null)
|
||||
_onQueueChanged?.Invoke(snapshot);
|
||||
}
|
||||
|
||||
public async Task InterruptAsync(CancellationToken ct)
|
||||
{
|
||||
await _sendLock.WaitAsync(ct);
|
||||
try
|
||||
{
|
||||
if (!_isTurnInFlight) return;
|
||||
|
||||
var requestId = Guid.NewGuid().ToString();
|
||||
var payload = JsonSerializer.Serialize(new
|
||||
{
|
||||
type = "control_request",
|
||||
request_id = requestId,
|
||||
request = new { subtype = "interrupt" }
|
||||
});
|
||||
|
||||
try { await _transport.WriteLineAsync(payload, ct); }
|
||||
catch (Exception ex) { _logger.LogWarning(ex, "Failed to write interrupt control_request; degrading gracefully."); }
|
||||
}
|
||||
finally
|
||||
{
|
||||
_sendLock.Release();
|
||||
}
|
||||
}
|
||||
|
||||
private async Task SendTurnAsync(string text, CancellationToken ct)
|
||||
{
|
||||
_isTurnInFlight = true;
|
||||
|
||||
var payload = JsonSerializer.Serialize(new
|
||||
{
|
||||
type = "user",
|
||||
message = new
|
||||
{
|
||||
role = "user",
|
||||
content = new[]
|
||||
{
|
||||
new { type = "text", text }
|
||||
}
|
||||
},
|
||||
parent_tool_use_id = (string?)null
|
||||
});
|
||||
|
||||
await _transport.WriteLineAsync(payload, ct);
|
||||
}
|
||||
|
||||
public async Task StopAsync()
|
||||
{
|
||||
_transport.Kill();
|
||||
await _transport.WaitForExitAsync();
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
await StopAsync();
|
||||
await _transport.DisposeAsync();
|
||||
_sendLock.Dispose();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user