feat(worker): remove a queued interactive message
StreamingClaudeSession.RemoveQueuedAsync drops the first occurrence of a queued message from _pending and re-broadcasts the updated queue. Wired through InteractiveSessionService + WorkerHub.RemoveQueuedInteractiveMessage + IWorkerClient.RemoveQueuedInteractiveMessageAsync. Removal by text (first match) is robust to a turn flushing mid-click. Fakes + ILiveSession impls updated.
This commit is contained in:
@@ -52,6 +52,7 @@ public interface IWorkerClient : INotifyPropertyChanged
|
|||||||
/// <summary>Answer a question a running task raised via AskUser.</summary>
|
/// <summary>Answer a question a running task raised via AskUser.</summary>
|
||||||
Task AnswerTaskQuestionAsync(string taskId, string questionId, string answer);
|
Task AnswerTaskQuestionAsync(string taskId, string questionId, string answer);
|
||||||
Task SendInteractiveMessageAsync(string taskId, string text);
|
Task SendInteractiveMessageAsync(string taskId, string text);
|
||||||
|
Task RemoveQueuedInteractiveMessageAsync(string taskId, string text);
|
||||||
Task StopInteractiveSessionAsync(string taskId);
|
Task StopInteractiveSessionAsync(string taskId);
|
||||||
Task InterruptInteractiveSessionAsync(string taskId);
|
Task InterruptInteractiveSessionAsync(string taskId);
|
||||||
/// <summary>The question a running task is currently blocked on, if any (for re-attach).</summary>
|
/// <summary>The question a running task is currently blocked on, if any (for re-attach).</summary>
|
||||||
|
|||||||
@@ -309,6 +309,12 @@ public partial class WorkerClient : ObservableObject, IAsyncDisposable, IWorkerC
|
|||||||
catch { /* offline or session already ended */ }
|
catch { /* offline or session already ended */ }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public async Task RemoveQueuedInteractiveMessageAsync(string taskId, string text)
|
||||||
|
{
|
||||||
|
try { await _hub.InvokeAsync("RemoveQueuedInteractiveMessage", taskId, text); }
|
||||||
|
catch { /* offline or session already ended */ }
|
||||||
|
}
|
||||||
|
|
||||||
public async Task StopInteractiveSessionAsync(string taskId)
|
public async Task StopInteractiveSessionAsync(string taskId)
|
||||||
{
|
{
|
||||||
try { await _hub.InvokeAsync("StopInteractiveSession", taskId); }
|
try { await _hub.InvokeAsync("StopInteractiveSession", taskId); }
|
||||||
|
|||||||
@@ -584,6 +584,9 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
|||||||
public Task InterruptInteractiveSession(string taskId) =>
|
public Task InterruptInteractiveSession(string taskId) =>
|
||||||
_interactive.InterruptAsync(taskId, Context.ConnectionAborted);
|
_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)
|
public async Task<DiscardPlanningOutcome> DiscardPlanningSessionAsync(string taskId, bool dequeueQueuedChildren = false)
|
||||||
{
|
{
|
||||||
var outcome = await _planning.DiscardAsync(taskId, dequeueQueuedChildren, Context.ConnectionAborted);
|
var outcome = await _planning.DiscardAsync(taskId, dequeueQueuedChildren, Context.ConnectionAborted);
|
||||||
|
|||||||
@@ -127,6 +127,12 @@ public sealed class InteractiveSessionService
|
|||||||
await session.SendUserMessageAsync(text, ct);
|
await session.SendUserMessageAsync(text, ct);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public async Task RemoveQueuedAsync(string taskId, string text, CancellationToken ct)
|
||||||
|
{
|
||||||
|
if (_registry.TryGet(taskId, out var session))
|
||||||
|
await session.RemoveQueuedAsync(text, ct);
|
||||||
|
}
|
||||||
|
|
||||||
public async Task InterruptAsync(string taskId, CancellationToken ct)
|
public async Task InterruptAsync(string taskId, CancellationToken ct)
|
||||||
{
|
{
|
||||||
if (_registry.TryGet(taskId, out var session))
|
if (_registry.TryGet(taskId, out var session))
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ public interface ILiveSession : IAsyncDisposable
|
|||||||
{
|
{
|
||||||
bool IsTurnInFlight { get; }
|
bool IsTurnInFlight { get; }
|
||||||
Task SendUserMessageAsync(string text, CancellationToken ct);
|
Task SendUserMessageAsync(string text, CancellationToken ct);
|
||||||
|
Task RemoveQueuedAsync(string text, CancellationToken ct);
|
||||||
Task InterruptAsync(CancellationToken ct);
|
Task InterruptAsync(CancellationToken ct);
|
||||||
Task StopAsync();
|
Task StopAsync();
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -118,6 +118,35 @@ public sealed class StreamingClaudeSession : ILiveSession
|
|||||||
_onUserMessageSent?.Invoke(text);
|
_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)
|
public async Task InterruptAsync(CancellationToken ct)
|
||||||
{
|
{
|
||||||
await _sendLock.WaitAsync(ct);
|
await _sendLock.WaitAsync(ct);
|
||||||
|
|||||||
@@ -148,6 +148,12 @@ public abstract class StubWorkerClient : IWorkerClient
|
|||||||
SentInteractive.Add((taskId, text));
|
SentInteractive.Add((taskId, text));
|
||||||
return Task.CompletedTask;
|
return Task.CompletedTask;
|
||||||
}
|
}
|
||||||
|
public List<(string TaskId, string Text)> RemovedQueued { get; } = new();
|
||||||
|
public virtual Task RemoveQueuedInteractiveMessageAsync(string taskId, string text)
|
||||||
|
{
|
||||||
|
RemovedQueued.Add((taskId, text));
|
||||||
|
return Task.CompletedTask;
|
||||||
|
}
|
||||||
public List<string> StoppedInteractive { get; } = new();
|
public List<string> StoppedInteractive { get; } = new();
|
||||||
public virtual Task StopInteractiveSessionAsync(string taskId)
|
public virtual Task StopInteractiveSessionAsync(string taskId)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -289,6 +289,8 @@ internal sealed class FakeLiveSession : ILiveSession
|
|||||||
return Task.CompletedTask;
|
return Task.CompletedTask;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public Task RemoveQueuedAsync(string text, CancellationToken ct) => Task.CompletedTask;
|
||||||
|
|
||||||
public Task InterruptAsync(CancellationToken ct) => Task.CompletedTask;
|
public Task InterruptAsync(CancellationToken ct) => Task.CompletedTask;
|
||||||
|
|
||||||
public Task StopAsync()
|
public Task StopAsync()
|
||||||
|
|||||||
@@ -11,6 +11,7 @@ public sealed class LiveSessionRegistryTests
|
|||||||
public bool IsTurnInFlight => false;
|
public bool IsTurnInFlight => false;
|
||||||
|
|
||||||
public Task SendUserMessageAsync(string text, CancellationToken ct) => Task.CompletedTask;
|
public Task SendUserMessageAsync(string text, CancellationToken ct) => Task.CompletedTask;
|
||||||
|
public Task RemoveQueuedAsync(string text, CancellationToken ct) => Task.CompletedTask;
|
||||||
public Task InterruptAsync(CancellationToken ct) => Task.CompletedTask;
|
public Task InterruptAsync(CancellationToken ct) => Task.CompletedTask;
|
||||||
|
|
||||||
public Task StopAsync()
|
public Task StopAsync()
|
||||||
|
|||||||
@@ -403,4 +403,67 @@ public sealed class StreamingClaudeSessionTests
|
|||||||
|
|
||||||
await session.DisposeAsync();
|
await session.DisposeAsync();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ──────────────────────────────────────────────────────────────────────────
|
||||||
|
// RemoveQueuedAsync tests
|
||||||
|
// ──────────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task RemoveQueued_RemovesFirstOccurrence_SnapshotContainsOnlySecond_AndSecondDeliveredOnResult()
|
||||||
|
{
|
||||||
|
var transport = new FakeClaudeStreamTransport();
|
||||||
|
var queueChanges = new List<IReadOnlyList<string>>();
|
||||||
|
var sent = new List<string>();
|
||||||
|
var session = BuildWithCallbacks(transport, queueChanges, sent);
|
||||||
|
|
||||||
|
await session.StartAsync([], "/tmp", "first", CancellationToken.None);
|
||||||
|
sent.Clear();
|
||||||
|
|
||||||
|
// Enqueue two messages while turn is in flight.
|
||||||
|
await session.SendUserMessageAsync("alpha", CancellationToken.None);
|
||||||
|
await session.SendUserMessageAsync("beta", CancellationToken.None);
|
||||||
|
queueChanges.Clear();
|
||||||
|
|
||||||
|
// Remove "alpha" from the queue.
|
||||||
|
await session.RemoveQueuedAsync("alpha", CancellationToken.None);
|
||||||
|
|
||||||
|
// Snapshot emitted and contains only "beta".
|
||||||
|
Assert.Single(queueChanges);
|
||||||
|
Assert.Equal(new[] { "beta" }, queueChanges[0]);
|
||||||
|
|
||||||
|
// Push result → only "beta" is flushed, not "alpha".
|
||||||
|
sent.Clear();
|
||||||
|
queueChanges.Clear();
|
||||||
|
await transport.PushLineAsync(ResultLine());
|
||||||
|
|
||||||
|
Assert.Single(sent);
|
||||||
|
Assert.Equal("beta", sent[0]);
|
||||||
|
|
||||||
|
// Queue now empty; next result leaves us idle.
|
||||||
|
await transport.PushLineAsync(ResultLine());
|
||||||
|
Assert.False(session.IsTurnInFlight);
|
||||||
|
|
||||||
|
await session.DisposeAsync();
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task RemoveQueued_NotFound_NoQueueChangedCallback()
|
||||||
|
{
|
||||||
|
var transport = new FakeClaudeStreamTransport();
|
||||||
|
var queueChanges = new List<IReadOnlyList<string>>();
|
||||||
|
var sent = new List<string>();
|
||||||
|
var session = BuildWithCallbacks(transport, queueChanges, sent);
|
||||||
|
|
||||||
|
await session.StartAsync([], "/tmp", "first", CancellationToken.None);
|
||||||
|
await session.SendUserMessageAsync("alpha", CancellationToken.None);
|
||||||
|
queueChanges.Clear();
|
||||||
|
|
||||||
|
// Try to remove a message that is not in the queue.
|
||||||
|
await session.RemoveQueuedAsync("nope", CancellationToken.None);
|
||||||
|
|
||||||
|
// No new snapshot emitted.
|
||||||
|
Assert.Empty(queueChanges);
|
||||||
|
|
||||||
|
await session.DisposeAsync();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -127,6 +127,7 @@ sealed class FakeWorkerClient : IWorkerClient
|
|||||||
public Task SetOnlineInboxAuthAsync(string refreshToken) => Task.CompletedTask;
|
public Task SetOnlineInboxAuthAsync(string refreshToken) => Task.CompletedTask;
|
||||||
public Task ClearOnlineInboxAuthAsync() => Task.CompletedTask;
|
public Task ClearOnlineInboxAuthAsync() => Task.CompletedTask;
|
||||||
public Task SendInteractiveMessageAsync(string taskId, string text) => Task.CompletedTask;
|
public Task SendInteractiveMessageAsync(string taskId, string text) => Task.CompletedTask;
|
||||||
|
public Task RemoveQueuedInteractiveMessageAsync(string taskId, string text) => Task.CompletedTask;
|
||||||
public Task StopInteractiveSessionAsync(string taskId) => Task.CompletedTask;
|
public Task StopInteractiveSessionAsync(string taskId) => Task.CompletedTask;
|
||||||
public Task InterruptInteractiveSessionAsync(string taskId) => Task.CompletedTask;
|
public Task InterruptInteractiveSessionAsync(string taskId) => Task.CompletedTask;
|
||||||
public IReadOnlyList<ActiveTask> GetActiveTasks() => System.Array.Empty<ActiveTask>();
|
public IReadOnlyList<ActiveTask> GetActiveTasks() => System.Array.Empty<ActiveTask>();
|
||||||
|
|||||||
Reference in New Issue
Block a user