diff --git a/Directory.Packages.props b/Directory.Packages.props index e7ac10898..5253d4491 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -13,6 +13,8 @@ + + @@ -37,4 +39,4 @@ - \ No newline at end of file + diff --git a/src/Netclaw.Actors.Tests/Cli/DaemonClientMappingTests.cs b/src/Netclaw.Actors.Tests/Cli/DaemonClientMappingTests.cs new file mode 100644 index 000000000..cabc61b90 --- /dev/null +++ b/src/Netclaw.Actors.Tests/Cli/DaemonClientMappingTests.cs @@ -0,0 +1,65 @@ +using Netclaw.Actors.Protocol; +using Netclaw.Cli.Daemon; +using Xunit; + +namespace Netclaw.Actors.Tests.Cli; + +public sealed class DaemonClientMappingTests +{ + [Fact] + public void FromDto_maps_text_delta_output() + { + var dto = new SessionOutputDto + { + Type = "text_delta", + SessionId = "signalr/test", + TimestampMs = 123, + Text = "hel" + }; + + var output = DaemonClient.FromDto(dto); + + var delta = Assert.IsType(output); + Assert.Equal("signalr/test", delta.SessionId.Value); + Assert.Equal("hel", delta.Delta); + } + + [Fact] + public void FromDto_maps_tool_result_output() + { + var dto = new SessionOutputDto + { + Type = "tool_result", + SessionId = "signalr/test", + TimestampMs = 123, + CallId = "abc", + ToolName = "bash", + Result = "ok" + }; + + var output = DaemonClient.FromDto(dto); + + var result = Assert.IsType(output); + Assert.Equal("signalr/test", result.SessionId.Value); + Assert.Equal("abc", result.CallId); + Assert.Equal("bash", result.ToolName); + Assert.Equal("ok", result.Result); + } + + [Fact] + public void FromDto_unknown_type_becomes_error_output() + { + var dto = new SessionOutputDto + { + Type = "mystery", + SessionId = "signalr/test", + TimestampMs = 123 + }; + + var output = DaemonClient.FromDto(dto); + + var error = Assert.IsType(output); + Assert.Contains("Unknown output type", error.Message); + Assert.Equal("signalr/test", error.SessionId.Value); + } +} diff --git a/src/Netclaw.Actors.Tests/Cli/DaemonClientReconnectIntegrationTests.cs b/src/Netclaw.Actors.Tests/Cli/DaemonClientReconnectIntegrationTests.cs new file mode 100644 index 000000000..2ba45d2d3 --- /dev/null +++ b/src/Netclaw.Actors.Tests/Cli/DaemonClientReconnectIntegrationTests.cs @@ -0,0 +1,175 @@ +using System.Collections.Concurrent; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Http.Connections; +using Microsoft.AspNetCore.SignalR; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Netclaw.Actors.Channels; +using Netclaw.Actors.Protocol; +using Netclaw.Cli.Daemon; +using Netclaw.Daemon.Gateway; +using Xunit; + +namespace Netclaw.Actors.Tests.Cli; + +public sealed class DaemonClientReconnectIntegrationTests +{ + [Fact] + public async Task EnsureSession_recreates_session_after_server_restart() + { + var port = GetFreeTcpPort(); + var host1 = await StartFakeHubAsync(port); + + await using var client = new DaemonClient($"http://127.0.0.1:{port}"); + var outputs = new List(); + var firstResponseReceived = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var secondResponseReceived = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + using var sub = client.SessionOutput.Subscribe(output => + { + outputs.Add(output); + + if (output is TextOutput { Text: "echo:first" }) + firstResponseReceived.TrySetResult(); + + if (output is TextOutput { Text: "echo:second" }) + secondResponseReceived.TrySetResult(); + }); + + await client.CreateSessionAsync("tui"); + await client.SendAsync(new ChannelInput + { + SenderId = "test", + Contents = [new TextContent("first")], + ReceivedAt = DateTimeOffset.UtcNow + }); + + await WaitFor(firstResponseReceived.Task, TimeSpan.FromSeconds(5)); + + await host1.StopAsync(); + host1.Dispose(); + + using var host2 = await StartFakeHubAsync(port); + + await client.EnsureSessionAsync("tui"); + await client.SendAsync(new ChannelInput + { + SenderId = "test", + Contents = [new TextContent("second")], + ReceivedAt = DateTimeOffset.UtcNow + }); + + await WaitFor(secondResponseReceived.Task, TimeSpan.FromSeconds(10)); + + var textOutputs = outputs.OfType().ToList(); + Assert.Contains(textOutputs, o => o.Text == "echo:first"); + Assert.Contains(textOutputs, o => o.Text == "echo:second"); + Assert.Contains(outputs, o => o is TurnCompleted); + } + + private static async Task WaitFor(Task task, TimeSpan timeout) + { + await task.WaitAsync(timeout); + } + + private static async Task StartFakeHubAsync(int port) + { + var builder = WebApplication.CreateBuilder(); + builder.WebHost.UseKestrel(); + builder.WebHost.UseUrls($"http://127.0.0.1:{port}"); + builder.Services.AddSignalR(); + builder.Services.AddSingleton(); + + var app = builder.Build(); + app.MapHub("/hub/session", options => + { + options.Transports = HttpTransportType.WebSockets | HttpTransportType.LongPolling; + }); + + await app.StartAsync(); + return app; + } + + private static int GetFreeTcpPort() + { + var listener = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0); + listener.Start(); + var port = ((System.Net.IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } + + private sealed class FakeHubState + { + private readonly object _gate = new(); + private readonly HashSet _sessions = []; + private readonly ConcurrentDictionary _connectionSessions = new(); + + public SessionEnsureResultDto Ensure(string connectionId, string? sessionId) + { + lock (_gate) + { + if (!string.IsNullOrWhiteSpace(sessionId) && _sessions.Contains(sessionId)) + { + _connectionSessions[connectionId] = sessionId; + return new SessionEnsureResultDto { SessionId = sessionId, Created = false }; + } + + var created = $"signalr/{Guid.NewGuid():N}"; + _sessions.Add(created); + _connectionSessions[connectionId] = created; + return new SessionEnsureResultDto { SessionId = created, Created = true }; + } + } + + public bool IsAttached(string connectionId, string sessionId) + => _connectionSessions.TryGetValue(connectionId, out var attached) + && string.Equals(attached, sessionId, StringComparison.Ordinal); + + public void Disconnect(string connectionId) + => _connectionSessions.TryRemove(connectionId, out _); + } + + private sealed class FakeSessionHub : Hub + { + private readonly FakeHubState _state; + + public FakeSessionHub(FakeHubState state) + { + _state = state; + } + + public Task EnsureSession(string? sessionId, string channelType) + => Task.FromResult(_state.Ensure(Context.ConnectionId, sessionId)); + + public async Task SendMessage(string sessionId, string text) + { + if (!_state.IsAttached(Context.ConnectionId, sessionId)) + throw new HubException("session not attached"); + + await Clients.Caller.ReceiveOutput(new SessionOutputDto + { + Type = "text", + SessionId = sessionId, + TimestampMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), + Text = $"echo:{text}" + }); + + await Clients.Caller.ReceiveOutput(new SessionOutputDto + { + Type = "turn_completed", + SessionId = sessionId, + TimestampMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), + TurnNumber = 1 + }); + } + + public override Task OnDisconnectedAsync(Exception? exception) + { + _state.Disconnect(Context.ConnectionId); + return base.OnDisconnectedAsync(exception); + } + } +} diff --git a/src/Netclaw.Actors.Tests/Netclaw.Actors.Tests.csproj b/src/Netclaw.Actors.Tests/Netclaw.Actors.Tests.csproj index 5018aeb02..b99d9695c 100644 --- a/src/Netclaw.Actors.Tests/Netclaw.Actors.Tests.csproj +++ b/src/Netclaw.Actors.Tests/Netclaw.Actors.Tests.csproj @@ -11,6 +11,7 @@ + diff --git a/src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs b/src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs index 1a5c56547..f1a9cc63c 100644 --- a/src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs @@ -434,7 +434,20 @@ public IAsyncEnumerable GetStreamingResponseAsync( ChatOptions? options = null, CancellationToken cancellationToken = default) { - throw new NotSupportedException("Streaming not used in tests"); + return CreateStreamingUpdatesAsync(messages, options, cancellationToken); + } + + private async IAsyncEnumerable CreateStreamingUpdatesAsync( + IEnumerable messages, + ChatOptions? options, + [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken) + { + var response = await GetResponseAsync(messages, options, cancellationToken); + foreach (var update in response.ToChatResponseUpdates()) + { + cancellationToken.ThrowIfCancellationRequested(); + yield return update; + } } public object? GetService(Type serviceType, object? serviceKey = null) => null; diff --git a/src/Netclaw.Actors/Protocol/SessionEnsureResultDto.cs b/src/Netclaw.Actors/Protocol/SessionEnsureResultDto.cs new file mode 100644 index 000000000..e4052d952 --- /dev/null +++ b/src/Netclaw.Actors/Protocol/SessionEnsureResultDto.cs @@ -0,0 +1,14 @@ +namespace Netclaw.Actors.Protocol; + +/// +/// Wire-safe response for ensuring a SignalR session binding. +/// +public sealed record SessionEnsureResultDto +{ + public required string SessionId { get; init; } + + /// + /// True when a new session was created; false when existing session was reattached. + /// + public required bool Created { get; init; } +} diff --git a/src/Netclaw.Actors/Protocol/SessionOutput.cs b/src/Netclaw.Actors/Protocol/SessionOutput.cs index 1e9481591..3b8d15e96 100644 --- a/src/Netclaw.Actors/Protocol/SessionOutput.cs +++ b/src/Netclaw.Actors/Protocol/SessionOutput.cs @@ -28,6 +28,15 @@ public sealed record TextOutput : SessionOutput public required string Text { get; init; } } +/// +/// Incremental text delta from the assistant while a turn is streaming. +/// Requires . +/// +public sealed record TextDeltaOutput : SessionOutput +{ + public required string Delta { get; init; } +} + /// /// Thinking/reasoning tokens from the model (e.g., Claude extended thinking). /// Requires . @@ -37,6 +46,15 @@ public sealed record ThinkingOutput : SessionOutput public required string Text { get; init; } } +/// +/// Incremental thinking/reasoning delta while a turn is streaming. +/// Requires . +/// +public sealed record ThinkingDeltaOutput : SessionOutput +{ + public required string Delta { get; init; } +} + /// /// The model has requested a tool/function call. /// Requires . diff --git a/src/Netclaw.Actors/Protocol/SessionOutputDto.cs b/src/Netclaw.Actors/Protocol/SessionOutputDto.cs index 0edad358e..b4868c01b 100644 --- a/src/Netclaw.Actors/Protocol/SessionOutputDto.cs +++ b/src/Netclaw.Actors/Protocol/SessionOutputDto.cs @@ -8,7 +8,7 @@ namespace Netclaw.Actors.Protocol; public sealed record SessionOutputDto { /// - /// Output type discriminator (e.g. "text", "thinking", "tool_call", + /// Output type discriminator (e.g. "text", "text_delta", "thinking", "thinking_delta", "tool_call", /// "tool_result", "usage", "turn_completed", "error", "compaction", /// "session_joined", "session_title"). /// diff --git a/src/Netclaw.Actors/Sessions/LlmMessages.cs b/src/Netclaw.Actors/Sessions/LlmMessages.cs index 86254b973..6e039b1e7 100644 --- a/src/Netclaw.Actors/Sessions/LlmMessages.cs +++ b/src/Netclaw.Actors/Sessions/LlmMessages.cs @@ -8,6 +8,18 @@ namespace Netclaw.Actors.Sessions; internal sealed record LlmResponseReceived { public required ChatResponse Response { get; init; } + + public bool StreamedText { get; init; } + + public bool StreamedThinking { get; init; } +} + +/// +/// Incremental streaming delta emitted while an LLM response is in-flight. +/// +internal sealed record LlmResponseDeltaReceived +{ + public required AIContent Content { get; init; } } /// diff --git a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs index 2db78de6d..9b6b604b2 100644 --- a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs +++ b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs @@ -1,4 +1,5 @@ using System.Diagnostics; +using System.Text; using System.Text.Json; using Akka.Actor; using Akka.Event; @@ -168,7 +169,29 @@ private void Processing() } // Normal text response — persist turn - HandleTextResponse(lastMessage, response.Usage); + HandleTextResponse(lastMessage, response.Usage, msg.StreamedText, msg.StreamedThinking); + }); + + Command(msg => + { + switch (msg.Content) + { + case TextContent text when !string.IsNullOrEmpty(text.Text): + EmitOutput(new TextDeltaOutput + { + SessionId = _sessionId, + Delta = text.Text + }, OutputFilter.Text); + break; + + case TextReasoningContent thinking when !string.IsNullOrEmpty(thinking.Text): + EmitOutput(new ThinkingDeltaOutput + { + SessionId = _sessionId, + Delta = thinking.Text + }, OutputFilter.Thinking); + break; + } }); Command(msg => @@ -468,7 +491,11 @@ private void HandleToolCallResponse( _ = ExecuteToolsAsync(executor, toolCalls, sessionId, auditLogger, tp, self); } - private void HandleTextResponse(AiChatMessage lastMessage, UsageDetails? usage) + private void HandleTextResponse( + AiChatMessage lastMessage, + UsageDetails? usage, + bool streamedText, + bool streamedThinking) { _toolIterationCount = 0; // Reset for potential buffer drain (new logical turn) @@ -501,7 +528,7 @@ private void HandleTextResponse(AiChatMessage lastMessage, UsageDetails? usage) TurnCount = _state.TurnCount + 1 }; - EmitResponseOutputs(lastMessage, usage); + EmitResponseOutputs(lastMessage, usage, includeText: true, includeThinking: true); MaybeSnapshot(); // Check if compaction should trigger @@ -638,8 +665,8 @@ private static async Task InvokeLlmAsync( { try { - var response = await client.GetResponseAsync(messages, options); - self.Tell(new LlmResponseReceived { Response = response }); + var response = await InvokeStreamingResponseAsync(client, messages, options, self); + self.Tell(response); } catch (Exception ex) { @@ -647,6 +674,103 @@ private static async Task InvokeLlmAsync( } } + private static async Task InvokeStreamingResponseAsync( + IChatClient client, + List messages, + ChatOptions? options, + IActorRef self) + { + var contents = new List(); + var updates = new List(); + var textBuilder = new StringBuilder(); + var thinkingBuilder = new StringBuilder(); + string? pendingTextDelta = null; + string? pendingThinkingDelta = null; + var textDeltaCount = 0; + var thinkingDeltaCount = 0; + + await foreach (var update in client.GetStreamingResponseAsync(messages, options)) + { + updates.Add(update); + + if (update.Contents is not null) + { + foreach (var content in update.Contents) + { + switch (content) + { + case TextContent text when !string.IsNullOrEmpty(text.Text): + textBuilder.Append(text.Text); + textDeltaCount++; + if (textDeltaCount == 1) + { + pendingTextDelta = text.Text; + } + else + { + if (textDeltaCount == 2 && !string.IsNullOrEmpty(pendingTextDelta)) + { + self.Tell(new LlmResponseDeltaReceived + { + Content = new TextContent(pendingTextDelta) + }); + } + + self.Tell(new LlmResponseDeltaReceived { Content = content }); + } + break; + + case TextReasoningContent thinking when !string.IsNullOrEmpty(thinking.Text): + thinkingBuilder.Append(thinking.Text); + thinkingDeltaCount++; + if (thinkingDeltaCount == 1) + { + pendingThinkingDelta = thinking.Text; + } + else + { + if (thinkingDeltaCount == 2 && !string.IsNullOrEmpty(pendingThinkingDelta)) + { + self.Tell(new LlmResponseDeltaReceived + { + Content = new TextReasoningContent(pendingThinkingDelta) + }); + } + + self.Tell(new LlmResponseDeltaReceived { Content = content }); + } + break; + + case FunctionCallContent: + contents.Add(content); + break; + } + } + } + + } + + if (thinkingBuilder.Length > 0) + contents.Add(new TextReasoningContent(thinkingBuilder.ToString())); + + if (textBuilder.Length > 0) + contents.Add(new TextContent(textBuilder.ToString())); + + var response = updates.Count > 0 + ? updates.ToChatResponse() + : new ChatResponse(new AiChatMessage(Microsoft.Extensions.AI.ChatRole.Assistant, contents)); + + if (response.Messages.Count == 0) + response.Messages.Add(new AiChatMessage(Microsoft.Extensions.AI.ChatRole.Assistant, contents)); + + return new LlmResponseReceived + { + Response = response, + StreamedText = textDeltaCount > 1, + StreamedThinking = thinkingDeltaCount > 1 + }; + } + private static async Task ExecuteToolsAsync( IToolExecutor executor, List toolCalls, @@ -726,13 +850,17 @@ private void MaybeSnapshot() } } - private void EmitResponseOutputs(AiChatMessage message, UsageDetails? usage) + private void EmitResponseOutputs( + AiChatMessage message, + UsageDetails? usage, + bool includeText = true, + bool includeThinking = true) { foreach (var content in message.Contents) { switch (content) { - case TextContent text: + case TextContent text when includeText: EmitOutput(new TextOutput { SessionId = _sessionId, @@ -740,7 +868,7 @@ private void EmitResponseOutputs(AiChatMessage message, UsageDetails? usage) }, OutputFilter.Text); break; - case TextReasoningContent thinking: + case TextReasoningContent thinking when includeThinking: EmitOutput(new ThinkingOutput { SessionId = _sessionId, diff --git a/src/Netclaw.Cli/Configuration/ChatClientFactory.cs b/src/Netclaw.Cli/Configuration/ChatClientFactory.cs deleted file mode 100644 index 7bbc01354..000000000 --- a/src/Netclaw.Cli/Configuration/ChatClientFactory.cs +++ /dev/null @@ -1,33 +0,0 @@ -using Microsoft.Extensions.AI; -using Netclaw.Configuration; -using OllamaSharp; - -namespace Netclaw.Cli.Configuration; - -/// -/// Transitional: duplicated from Netclaw.Daemon. Removed in Task 1.28 -/// when CLI connects to daemon via SignalR instead of running in-process. -/// -internal sealed class ChatClientFactory -{ - private readonly Dictionary _providers; - - public ChatClientFactory(Dictionary providers) - => _providers = providers; - - public IChatClient Create(ModelReference model) - { - if (!_providers.TryGetValue(model.Provider, out var provider)) - throw new InvalidOperationException( - $"Provider '{model.Provider}' not found. " - + $"Configured: {string.Join(", ", _providers.Keys)}"); - - return provider.Type.ToLowerInvariant() switch - { - "ollama" => new OllamaApiClient( - new Uri(provider.Endpoint), model.ModelId), - _ => throw new InvalidOperationException( - $"Unknown provider type '{provider.Type}'. Supported: ollama") - }; - } -} diff --git a/src/Netclaw.Cli/Configuration/NetclawChatClientProvider.cs b/src/Netclaw.Cli/Configuration/NetclawChatClientProvider.cs deleted file mode 100644 index 7e467b078..000000000 --- a/src/Netclaw.Cli/Configuration/NetclawChatClientProvider.cs +++ /dev/null @@ -1,31 +0,0 @@ -using Microsoft.Extensions.AI; -using Netclaw.Configuration; - -namespace Netclaw.Cli.Configuration; - -/// -/// Transitional: duplicated from Netclaw.Daemon. Removed in Task 1.28 -/// when CLI connects to daemon via SignalR instead of running in-process. -/// -internal sealed class NetclawChatClientProvider : IChatClientProvider -{ - private readonly IChatClient _main; - private readonly IChatClient? _fallback; - private readonly IChatClient? _compaction; - - public NetclawChatClientProvider(ChatClientFactory factory, ModelSelection models) - { - _main = factory.Create(models.Main); - _fallback = models.Fallback is not null - ? factory.Create(models.Fallback) : null; - _compaction = models.Compaction is not null - ? factory.Create(models.Compaction) : null; - } - - public IChatClient GetClient(ModelRole role) => role switch - { - ModelRole.Fallback => _fallback ?? _main, - ModelRole.Compaction => _compaction ?? _main, - _ => _main - }; -} diff --git a/src/Netclaw.Cli/Daemon/DaemonClient.cs b/src/Netclaw.Cli/Daemon/DaemonClient.cs new file mode 100644 index 000000000..40f3d2927 --- /dev/null +++ b/src/Netclaw.Cli/Daemon/DaemonClient.cs @@ -0,0 +1,397 @@ +using System.Reactive.Linq; +using System.Reactive.Subjects; +using Microsoft.AspNetCore.SignalR.Client; +using Microsoft.Extensions.AI; +using Netclaw.Actors.Channels; +using Netclaw.Actors.Protocol; + +namespace Netclaw.Cli.Daemon; + +/// +/// Thin SignalR client for daemon-backed sessions. +/// Maintains connection state, session attachment across reconnects, +/// and exposes mapped events for the TUI. +/// +public sealed class DaemonClient : IAsyncDisposable +{ + private static readonly TimeSpan[] ReconnectDelays = + [ + TimeSpan.Zero, + TimeSpan.FromSeconds(2), + TimeSpan.FromSeconds(5), + TimeSpan.FromSeconds(10) + ]; + + private readonly HubConnection _connection; + private readonly string _daemonEndpoint; + private readonly string _hubUrl; + private readonly Subject _outputSubject = new(); + private readonly Subject _connectionSubject = new(); + private readonly SemaphoreSlim _connectGate = new(1, 1); + private readonly CancellationTokenSource _lifetimeCts = new(); + + private string? _sessionId; + private string? _channelType; + private bool _hasConnected; + private bool _disposed; + + public DaemonClient(string daemonEndpoint) + { + if (string.IsNullOrWhiteSpace(daemonEndpoint)) + throw new ArgumentException("Daemon endpoint cannot be empty.", nameof(daemonEndpoint)); + + _daemonEndpoint = daemonEndpoint.TrimEnd('/'); + _hubUrl = BuildHubUrl(_daemonEndpoint); + + _connection = new HubConnectionBuilder() + .WithUrl(_hubUrl) + .WithAutomaticReconnect(ReconnectDelays) + .Build(); + + _connection.On("ReceiveOutput", dto => + { + _outputSubject.OnNext(FromDto(dto)); + }); + + _connection.Reconnected += async _ => + { + if (!string.IsNullOrWhiteSpace(_channelType)) + await EnsureSessionInternalAsync(_channelType!, CancellationToken.None); + + _connectionSubject.OnNext(new DaemonConnectionEvent( + DaemonConnectionState.Connected, + _daemonEndpoint, + $"Reconnected to daemon at {_daemonEndpoint}.")); + }; + + _connection.Reconnecting += ex => + { + var reason = ex?.Message ?? "connection dropped"; + _connectionSubject.OnNext(new DaemonConnectionEvent( + DaemonConnectionState.Reconnecting, + _daemonEndpoint, + $"Reconnecting to {_daemonEndpoint}: {reason}")); + return Task.CompletedTask; + }; + + _connection.Closed += async ex => + { + if (_disposed) + return; + + var reason = ex?.Message ?? "connection closed"; + _connectionSubject.OnNext(new DaemonConnectionEvent( + DaemonConnectionState.Disconnected, + _daemonEndpoint, + $"Disconnected from daemon at {_daemonEndpoint}: {reason}")); + + if (!string.IsNullOrWhiteSpace(_sessionId)) + await ReconnectLoopAsync(); + }; + } + + public IObservable SessionOutput => _outputSubject.AsObservable(); + public IObservable ConnectionEvents => _connectionSubject.AsObservable(); + + public bool IsConnected => _connection.State is HubConnectionState.Connected; + + public async Task ConnectAsync(CancellationToken cancellationToken = default) + { + if (IsConnected) + return; + + await _connectGate.WaitAsync(cancellationToken); + try + { + if (IsConnected) + return; + + // Another reconnect/start sequence may already be in-flight. + if (_connection.State is HubConnectionState.Connecting or HubConnectionState.Reconnecting) + { + await WaitForStableConnectionStateAsync(cancellationToken); + if (IsConnected) + return; + } + + _connectionSubject.OnNext(new DaemonConnectionEvent( + DaemonConnectionState.Connecting, + _daemonEndpoint, + $"Connecting to daemon at {_daemonEndpoint}...")); + + Exception? lastError = null; + foreach (var delay in ReconnectDelays) + { + if (delay > TimeSpan.Zero) + await Task.Delay(delay, cancellationToken); + + try + { + await _connection.StartAsync(cancellationToken); + + _connectionSubject.OnNext(new DaemonConnectionEvent( + DaemonConnectionState.Connected, + _daemonEndpoint, + _hasConnected + ? $"Reconnected to daemon at {_daemonEndpoint}." + : $"Connected to daemon at {_daemonEndpoint}.")); + _hasConnected = true; + return; + } + catch (Exception ex) + { + lastError = ex; + } + } + + throw new InvalidOperationException("Failed to connect to daemon SignalR hub.", lastError); + } + finally + { + _connectGate.Release(); + } + } + + private async Task WaitForStableConnectionStateAsync(CancellationToken cancellationToken) + { + var deadline = DateTimeOffset.UtcNow.AddSeconds(15); + while (DateTimeOffset.UtcNow < deadline) + { + if (_connection.State is HubConnectionState.Connected or HubConnectionState.Disconnected) + return; + + await Task.Delay(100, cancellationToken); + } + } + + public async Task CreateSessionAsync( + string channelType, + CancellationToken cancellationToken = default) + { + if (string.IsNullOrWhiteSpace(channelType)) + throw new ArgumentException("Channel type cannot be empty.", nameof(channelType)); + + _channelType = channelType; + _sessionId = null; + return await EnsureSessionInternalAsync(channelType, cancellationToken); + } + + public async Task EnsureSessionAsync( + string channelType, + CancellationToken cancellationToken = default) + { + _channelType = channelType; + return await EnsureSessionInternalAsync(channelType, cancellationToken); + } + + public async Task SendAsync(ChannelInput input, CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(input); + + await ConnectAsync(cancellationToken); + + var sessionId = _sessionId; + if (string.IsNullOrWhiteSpace(sessionId)) + throw new InvalidOperationException("Session not initialized. Call CreateSessionAsync first."); + + var text = input.Contents.OfType().Select(x => x.Text).FirstOrDefault(); + if (string.IsNullOrWhiteSpace(text)) + throw new InvalidOperationException("Only non-empty text messages are currently supported."); + + await _connection.InvokeCoreAsync( + "SendMessage", + [sessionId, text], + cancellationToken); + } + + public async ValueTask DisposeAsync() + { + _outputSubject.Dispose(); + _connectionSubject.Dispose(); + _lifetimeCts.Cancel(); + _lifetimeCts.Dispose(); + _disposed = true; + await _connection.DisposeAsync(); + _connectGate.Dispose(); + } + + private async Task ReconnectLoopAsync() + { + if (_disposed) + return; + + const int maxAttempts = 20; + var attempts = 0; + while (!_disposed && !_lifetimeCts.Token.IsCancellationRequested) + { + attempts++; + try + { + _connectionSubject.OnNext(new DaemonConnectionEvent( + DaemonConnectionState.Reconnecting, + _daemonEndpoint, + $"Retrying daemon connection at {_daemonEndpoint} (attempt {attempts}/{maxAttempts})...", + attempts, + maxAttempts, + 0)); + + await ConnectAsync(_lifetimeCts.Token); + return; + } + catch when (!_disposed && !_lifetimeCts.Token.IsCancellationRequested) + { + if (attempts >= maxAttempts) + { + _connectionSubject.OnNext(new DaemonConnectionEvent( + DaemonConnectionState.Disconnected, + _daemonEndpoint, + $"Unable to reconnect to daemon at {_daemonEndpoint} after {maxAttempts} attempts.", + attempts, + maxAttempts, + 0)); + return; + } + + for (var countdown = 2; countdown > 0; countdown--) + { + _connectionSubject.OnNext(new DaemonConnectionEvent( + DaemonConnectionState.Reconnecting, + _daemonEndpoint, + $"Retrying daemon connection at {_daemonEndpoint} (attempt {attempts + 1}/{maxAttempts}) in {countdown}s...", + attempts + 1, + maxAttempts, + countdown)); + + await Task.Delay(TimeSpan.FromSeconds(1), _lifetimeCts.Token); + } + } + } + } + + private static string BuildHubUrl(string endpoint) + { + var trimmed = endpoint.TrimEnd('/'); + return $"{trimmed}/hub/session"; + } + + private async Task EnsureSessionInternalAsync( + string channelType, + CancellationToken cancellationToken) + { + await ConnectAsync(cancellationToken); + + var result = await _connection.InvokeCoreAsync( + "EnsureSession", + [_sessionId, channelType], + cancellationToken); + + _sessionId = result.SessionId; + + if (result.Created) + { + _connectionSubject.OnNext(new DaemonConnectionEvent( + DaemonConnectionState.Connected, + _daemonEndpoint, + $"Created a new daemon session at {_daemonEndpoint}.")); + } + + return result.SessionId; + } + + internal static SessionOutput FromDto(SessionOutputDto dto) + { + var sessionId = new SessionId(dto.SessionId); + + return dto.Type switch + { + "text" => new TextOutput + { + SessionId = sessionId, + TimestampMs = dto.TimestampMs, + Text = dto.Text ?? string.Empty + }, + "text_delta" => new TextDeltaOutput + { + SessionId = sessionId, + TimestampMs = dto.TimestampMs, + Delta = dto.Text ?? string.Empty + }, + "thinking" => new ThinkingOutput + { + SessionId = sessionId, + TimestampMs = dto.TimestampMs, + Text = dto.Text ?? string.Empty + }, + "thinking_delta" => new ThinkingDeltaOutput + { + SessionId = sessionId, + TimestampMs = dto.TimestampMs, + Delta = dto.Text ?? string.Empty + }, + "tool_call" => new ToolCallOutput + { + SessionId = sessionId, + TimestampMs = dto.TimestampMs, + CallId = dto.CallId ?? string.Empty, + ToolName = dto.ToolName ?? "unknown", + ArgumentsJson = dto.ArgumentsJson + }, + "tool_result" => new ToolResultOutput + { + SessionId = sessionId, + TimestampMs = dto.TimestampMs, + CallId = dto.CallId ?? string.Empty, + ToolName = dto.ToolName ?? "unknown", + Result = dto.Result ?? string.Empty + }, + "usage" => new UsageOutput + { + SessionId = sessionId, + TimestampMs = dto.TimestampMs, + InputTokens = dto.InputTokens, + OutputTokens = dto.OutputTokens, + TotalTokens = dto.TotalTokens, + ContextWindowTokens = dto.ContextWindowTokens ?? 0, + UsagePercent = dto.UsagePercent + }, + "turn_completed" => new TurnCompleted + { + SessionId = sessionId, + TimestampMs = dto.TimestampMs, + TurnNumber = dto.TurnNumber ?? 0 + }, + "session_title" => new SessionTitleOutput + { + SessionId = sessionId, + TimestampMs = dto.TimestampMs, + Title = dto.Title ?? string.Empty + }, + "error" => new ErrorOutput + { + SessionId = sessionId, + TimestampMs = dto.TimestampMs, + Message = dto.ErrorMessage ?? "Unknown daemon error" + }, + "compaction" => new CompactionOutput + { + SessionId = sessionId, + TimestampMs = dto.TimestampMs, + MessagesBefore = dto.MessagesBefore ?? 0, + MessagesAfter = dto.MessagesAfter ?? 0 + }, + "session_joined" => new SessionJoined + { + SessionId = sessionId, + TimestampMs = dto.TimestampMs, + Title = dto.Title, + TurnCount = dto.TurnCount ?? 0 + }, + _ => new ErrorOutput + { + SessionId = sessionId, + TimestampMs = dto.TimestampMs, + Message = $"Unknown output type from daemon: {dto.Type}" + } + }; + } +} diff --git a/src/Netclaw.Cli/Daemon/DaemonConnectionEvent.cs b/src/Netclaw.Cli/Daemon/DaemonConnectionEvent.cs new file mode 100644 index 000000000..35a99e331 --- /dev/null +++ b/src/Netclaw.Cli/Daemon/DaemonConnectionEvent.cs @@ -0,0 +1,17 @@ +namespace Netclaw.Cli.Daemon; + +public enum DaemonConnectionState +{ + Connecting, + Connected, + Reconnecting, + Disconnected +} + +public sealed record DaemonConnectionEvent( + DaemonConnectionState State, + string Endpoint, + string Message, + int? Attempt = null, + int? MaxAttempts = null, + int? SecondsUntilRetry = null); diff --git a/src/Netclaw.Cli/Daemon/DaemonManager.cs b/src/Netclaw.Cli/Daemon/DaemonManager.cs index 70d049cc1..0e3046e91 100644 --- a/src/Netclaw.Cli/Daemon/DaemonManager.cs +++ b/src/Netclaw.Cli/Daemon/DaemonManager.cs @@ -110,8 +110,16 @@ public async Task StopAsync() // Wait up to 10 seconds for graceful exit. if (!await WaitForExitAsync(process, TimeSpan.FromSeconds(10))) { - // Timed out — force kill and wait briefly again. - TryKillProcess(process, out var killError); + // Timed out — hard cutoff. + string? killError = null; + if (OperatingSystem.IsLinux() || OperatingSystem.IsMacOS()) + { + if (!SendSignal(pid, Signal.SIGKILL)) + killError = "Failed to send SIGKILL."; + } + + if (!TryKillProcess(process, out var processKillError) && string.IsNullOrWhiteSpace(killError)) + killError = processKillError; if (!await WaitForExitAsync(process, TimeSpan.FromSeconds(5))) { @@ -439,8 +447,12 @@ private static async Task RunCommandAsync(string command, string a } } - // SIGTERM via P/Invoke — .NET's Process.Kill() sends SIGKILL - private enum Signal { SIGTERM = 15 } + // POSIX signals via P/Invoke + private enum Signal + { + SIGKILL = 9, + SIGTERM = 15 + } [LibraryImport("libc", SetLastError = true)] private static partial int kill(int pid, int sig); diff --git a/src/Netclaw.Cli/HeadlessChannel.cs b/src/Netclaw.Cli/HeadlessChannel.cs index ac8a77fda..ba3ecba71 100644 --- a/src/Netclaw.Cli/HeadlessChannel.cs +++ b/src/Netclaw.Cli/HeadlessChannel.cs @@ -1,13 +1,10 @@ -using Akka.Actor; -using Akka.Streams; -using Akka.Streams.Dsl; using Microsoft.Extensions.AI; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; -using Netclaw.Actors.Channels; using Netclaw.Configuration; using Netclaw.Actors.Protocol; using Netclaw.Channels; +using Netclaw.Cli.Daemon; namespace Netclaw.Cli; @@ -18,34 +15,33 @@ namespace Netclaw.Cli; /// public sealed class HeadlessChannel : IChannel { - private readonly SessionPipeline _pipeline; - private readonly ActorSystem _system; + private readonly DaemonClient _daemonClient; private readonly NetclawPaths _paths; private readonly IHostApplicationLifetime _lifetime; private readonly TimeProvider _timeProvider; private readonly string _prompt; private readonly ILogger _logger; - private MaterializedSession? _session; + private bool _isConnected; + private bool _receivedTextDeltaInCurrentTurn; + private bool _receivedThinkingDeltaInCurrentTurn; public string ChannelType => "headless"; public string DisplayName => "Headless Prompt"; - public ChannelHealth GetHealth() => _session is not null + public ChannelHealth GetHealth() => _isConnected ? new ChannelHealth(ChannelHealthStatus.Healthy) - : new ChannelHealth(ChannelHealthStatus.Disconnected, "No active session"); + : new ChannelHealth(ChannelHealthStatus.Disconnected, "No active daemon connection"); public HeadlessChannel( - SessionPipeline pipeline, - ActorSystem system, + DaemonClient daemonClient, NetclawPaths paths, IHostApplicationLifetime lifetime, TimeProvider timeProvider, string prompt, ILogger logger) { - _pipeline = pipeline; - _system = system; + _daemonClient = daemonClient; _paths = paths; _lifetime = lifetime; _timeProvider = timeProvider; @@ -61,8 +57,8 @@ public Task StartAsync(CancellationToken cancellationToken) public async Task StopAsync(CancellationToken cancellationToken) { - if (_session is not null) - await _session.DisposeAsync(); + _isConnected = false; + await Task.CompletedTask; } private async Task RunHeadlessAsync(CancellationToken stopping) @@ -75,43 +71,53 @@ private async Task RunHeadlessAsync(CancellationToken stopping) _paths.EnsureDirectoriesExist(); var logFileName = $"{sessionId.Value.Replace("/", "-")}.log"; var logPath = Path.Combine(_paths.LogsDirectory, logFileName); - var logWriter = new StreamWriter(logPath, append: false) { AutoFlush = true }; + await using var logWriter = new StreamWriter(logPath, append: false) { AutoFlush = true }; logWriter.WriteLine($"[{_timeProvider.GetUtcNow():o}] Headless session started: {sessionId}"); logWriter.WriteLine($"[{_timeProvider.GetUtcNow():o}] PROMPT: {_prompt}"); - // Create session pipeline - _session = await _pipeline.CreateAsync(sessionId, new SessionPipelineOptions + var turnCompleted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + using var connectionSubscription = _daemonClient.ConnectionEvents.Subscribe(evt => { - ChannelType = ChannelType - }, stopping); + Log(logWriter, $"CONNECTION: {evt.Message}"); + }); - // Materialize output stream → console + disk logging, exit on TurnCompleted - _session.Output - .To(Sink.ForEach(output => HandleOutput(output, logWriter))) - .Run(_system); + using var subscription = _daemonClient.SessionOutput.Subscribe(output => + { + HandleOutput(output, logWriter); + if (output is TurnCompleted) + turnCompleted.TrySetResult(); + }); + + await _daemonClient.ConnectAsync(stopping); + _isConnected = true; - // Materialize input with queue and send the single prompt - var inputQueue = Source.Queue(16, OverflowStrategy.Backpressure) - .ToMaterialized(_session.Input, Keep.Left) - .Run(_system); + sessionId = new SessionId(await _daemonClient.CreateSessionAsync(ChannelType, stopping)); - await inputQueue.OfferAsync(new ChannelInput + await _daemonClient.SendAsync(new Netclaw.Actors.Channels.ChannelInput { SenderId = "local-user", Contents = [new TextContent(_prompt)], ReceivedAt = _timeProvider.GetUtcNow() - }); + }, stopping); _logger.LogInformation("Headless session started: {SessionId} (log: {LogPath})", sessionId, logPath); + + await turnCompleted.Task.WaitAsync(stopping); + _lifetime.StopApplication(); } catch (OperationCanceledException ex) { _logger.LogDebug(ex, "Headless channel cancelled (shutdown)"); + WriteFailureLog("CANCELLED", ex); } catch (Exception ex) { _logger.LogError(ex, "Headless channel failed"); + Console.Error.WriteLine($"[headless:error] {ex.Message}"); + WriteFailureLog("FAILED", ex); + Environment.ExitCode = 1; _lifetime.StopApplication(); } } @@ -125,15 +131,39 @@ private void HandleOutput(SessionOutput output, StreamWriter log) break; case TextOutput msg: + if (_receivedTextDeltaInCurrentTurn) + { + Log(log, $"ASSISTANT_FINAL: {msg.Text}"); + break; + } + Console.WriteLine(msg.Text); Log(log, $"ASSISTANT: {msg.Text}"); break; + case TextDeltaOutput msg: + _receivedTextDeltaInCurrentTurn = true; + Console.Write(msg.Delta); + Log(log, $"ASSISTANT_DELTA: {msg.Delta}"); + break; + case ThinkingOutput msg: + if (_receivedThinkingDeltaInCurrentTurn) + { + Log(log, $"THINKING_FINAL: {msg.Text}"); + break; + } + Console.WriteLine($"[thinking] {msg.Text}"); Log(log, $"THINKING: {msg.Text}"); break; + case ThinkingDeltaOutput msg: + _receivedThinkingDeltaInCurrentTurn = true; + Console.Write($"[thinking]{msg.Delta}"); + Log(log, $"THINKING_DELTA: {msg.Delta}"); + break; + case ToolCallOutput msg: Console.WriteLine($"[tool:call] {msg.ToolName}({msg.ArgumentsJson ?? ""})"); Log(log, $"TOOL_CALL: {msg.ToolName} call_id={msg.CallId} args={msg.ArgumentsJson ?? "{}"}"); @@ -157,9 +187,11 @@ private void HandleOutput(SessionOutput output, StreamWriter log) break; case TurnCompleted msg: + Console.WriteLine(); Log(log, $"TURN_COMPLETED: turn={msg.TurnNumber}"); Log(log, "SESSION_ENDED"); - _lifetime.StopApplication(); + _receivedTextDeltaInCurrentTurn = false; + _receivedThinkingDeltaInCurrentTurn = false; break; case CompactionOutput msg: @@ -173,4 +205,19 @@ private void Log(StreamWriter log, string message) { log.WriteLine($"[{_timeProvider.GetUtcNow():o}] {message}"); } + + private void WriteFailureLog(string kind, Exception ex) + { + try + { + _paths.EnsureDirectoriesExist(); + var path = Path.Combine(_paths.LogsDirectory, "headless-errors.log"); + File.AppendAllText(path, + $"[{_timeProvider.GetUtcNow():o}] {kind}: {ex}\n"); + } + catch (Exception logEx) + { + Console.Error.WriteLine($"[headless:error] Failed to write failure log: {logEx.Message}"); + } + } } diff --git a/src/Netclaw.Cli/Netclaw.Cli.csproj b/src/Netclaw.Cli/Netclaw.Cli.csproj index eb44ed01a..1e52a1581 100644 --- a/src/Netclaw.Cli/Netclaw.Cli.csproj +++ b/src/Netclaw.Cli/Netclaw.Cli.csproj @@ -1,4 +1,4 @@ - + @@ -12,14 +12,10 @@ - - - - + - diff --git a/src/Netclaw.Cli/Program.cs b/src/Netclaw.Cli/Program.cs index ab60949df..1c974ba68 100644 --- a/src/Netclaw.Cli/Program.cs +++ b/src/Netclaw.Cli/Program.cs @@ -1,15 +1,9 @@ -using Akka.Hosting; -using Akka.Persistence.Hosting; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; -using Netclaw.Actors.Channels; -using Netclaw.Actors.Hosting; -using Netclaw.Actors.Tools; using Netclaw.Channels; using Netclaw.Cli; -using Netclaw.Cli.Configuration; using Netclaw.Cli.Daemon; using Netclaw.Cli.Tui; using Netclaw.Configuration; @@ -106,24 +100,17 @@ static async Task RunAsync(string[] args) return; } - // ── Interactive / headless modes (transitional: full Akka stack in-process) ── - // Task 1.28 refactors these to connect to daemon via SignalR instead. + // ── Interactive / headless modes (daemon-backed via SignalR) ── var webBuilder = WebApplication.CreateBuilder(args); - - // Transitional: use port 0 (random) to avoid conflict with daemon on 5199. - // Removed in Task 1.28 when CLI switches to SignalR client. webBuilder.WebHost.UseUrls("http://127.0.0.1:0"); var sharedPaths = ConfigureConfigServices(webBuilder.Services, webBuilder.Configuration); - ConfigureDaemonServices(webBuilder.Services, webBuilder.Configuration, sharedPaths); + ConfigureCliChatServices(webBuilder.Services, webBuilder.Configuration); // Suppress framework console logging — console is reserved for the chat UI webBuilder.Logging.ClearProviders(); webBuilder.Logging.SetMinimumLevel(LogLevel.Warning); - // SignalR (transitional — needed for in-process service stack) - webBuilder.Services.AddSignalR(); - // Channel selection based on mode switch (mode) { @@ -211,25 +198,14 @@ static NetclawPaths ConfigureConfigServices(IServiceCollection services, IConfig // TimeProvider (virtualized for testing) services.AddSingleton(TimeProvider.System); - // Providers and model resolution - var providers = configuration.GetSection("Providers") - .Get>() - ?? new() { ["local-ollama"] = new ProviderEntry() }; - var models = configuration.GetSection("Models") - .Get() ?? new ModelSelection(); - - var factory = new ChatClientFactory(providers); - var clientProvider = new NetclawChatClientProvider(factory, models); - services.AddSingleton(clientProvider); - return paths; } // ═══════════════════════════════════════════════════════════════════════ -// Transitional daemon services (removed in Task 1.28 when CLI uses SignalR) +// Daemon-backed CLI services (SignalR thin client) // ═══════════════════════════════════════════════════════════════════════ -static void ConfigureDaemonServices(IServiceCollection services, IConfigurationManager configuration, NetclawPaths paths) +static void ConfigureCliChatServices(IServiceCollection services, IConfigurationManager configuration) { // Resolve models for session config var models = configuration.GetSection("Models") @@ -248,39 +224,9 @@ static void ConfigureDaemonServices(IServiceCollection services, IConfigurationM MaxToolIterationsPerTurn = sessionSection.GetValue("MaxToolIterationsPerTurn", 10), }); - // Tools (auto-bound, no required properties) - var toolConfig = configuration.GetSection("Tools") - .Get() ?? new ToolConfig(); - services.AddSingleton(toolConfig); - - var toolRegistry = new ToolRegistry(); - toolRegistry.WithFirstPartyTools(toolConfig); - services.AddSingleton(toolRegistry); - services.AddSingleton(new DispatchingToolExecutor(toolRegistry)); - - // System prompt (file-based, with first-run seed) - if (!File.Exists(paths.PersonalityPath)) - File.WriteAllText(paths.PersonalityPath, - "You are Netclaw, a helpful homelab operations assistant. " - + "Be concise and direct."); - services.AddSingleton( - new FileSystemPromptProvider(paths)); - - // Akka.NET actor system - services.AddAkka("netclaw", (akkaBuilder, sp) => - { - akkaBuilder - .ConfigureLoggers(setup => - { - setup.ClearLoggers(); - setup.AddLoggerFactory(); - setup.LogLevel = Akka.Event.LogLevel.WarningLevel; - }) - .WithInMemoryJournal() - .WithInMemorySnapshotStore() - .WithNetclawActors(); - }); - - // Session pipeline (stream API for channels) - services.AddSingleton(); + var daemonEndpoint = + configuration["Daemon:Endpoint"] + ?? Environment.GetEnvironmentVariable("NETCLAW_DAEMON_ENDPOINT") + ?? "http://127.0.0.1:5199"; + services.AddSingleton(new DaemonClient(daemonEndpoint)); } diff --git a/src/Netclaw.Cli/Tui/ChatPage.cs b/src/Netclaw.Cli/Tui/ChatPage.cs index 46115af20..4afc2f182 100644 --- a/src/Netclaw.Cli/Tui/ChatPage.cs +++ b/src/Netclaw.Cli/Tui/ChatPage.cs @@ -1,5 +1,6 @@ using System.Reactive.Disposables; using System.Reactive.Linq; +using System.Text; using Netclaw.Actors.Protocol; using Termina.Components.Streaming; using Termina.Extensions; @@ -30,6 +31,10 @@ public sealed class ChatPage : ReactivePage // Track active tool timer so we can read final elapsed on completion private ElapsedTimeSegment? _toolTimer; + // Track active streamed assistant text segment + private SegmentId _assistantSegmentId; + private readonly StringBuilder _assistantBuffer = new(); + protected override void OnBound() { base.OnBound(); @@ -113,7 +118,11 @@ private LayoutNode BuildStatusBar() var barColor = status switch { "Ready" => Color.Green, - "Connecting..." => Color.Yellow, + _ when status.StartsWith("Connecting", StringComparison.Ordinal) => Color.Yellow, + _ when status.StartsWith("Reconnecting", StringComparison.Ordinal) => Color.Yellow, + _ when status.StartsWith("Connected", StringComparison.Ordinal) => Color.Green, + _ when status.StartsWith("Reconnected", StringComparison.Ordinal) => Color.Green, + _ when status.StartsWith("Disconnected", StringComparison.Ordinal) => Color.Red, _ when status.StartsWith("Generating") => Color.Yellow, _ when status.StartsWith("Connection failed") => Color.Red, _ => Color.BrightBlack @@ -173,12 +182,31 @@ private void HandleOutput(SessionOutput output) case TextOutput msg: // Remove thinking spinner if present RemoveThinkingSpinner(); - _chatHistory.AppendLine(""); + + // When streaming deltas were already rendered, TextOutput is the + // final full snapshot for compatibility. Finalize without duplicating. + if (_assistantSegmentId.Value != 0) + { + FinalizeAssistantSegmentIfNeeded(); + _chatHistory.ScrollToBottom(); + break; + } + _chatHistory.AppendLine($"Netclaw: {msg.Text}", Color.White); _chatHistory.AppendLine(""); _chatHistory.ScrollToBottom(); break; + case TextDeltaOutput msg: + RemoveThinkingSpinner(); + EnsureAssistantSegment(); + _assistantBuffer.Append(msg.Delta); + _chatHistory.Replace(_assistantSegmentId, + new StaticTextSegment($"Netclaw: {_assistantBuffer}", Color.White), + keepTracked: true); + _chatHistory.ScrollToBottom(); + break; + case ThinkingOutput: // Hidden — reasoning output is too verbose for the chat view. // TODO: collapsible thinking sections when Termina supports it. @@ -237,6 +265,7 @@ private void HandleOutput(SessionOutput output) case TurnCompleted: RemoveThinkingSpinner(); + FinalizeAssistantSegmentIfNeeded(); ViewModel.StatusMessage = "Ready"; _chatHistory.ScrollToBottom(); break; @@ -270,4 +299,28 @@ private static string FormatElapsed(TimeSpan elapsed) => elapsed.TotalSeconds < 60 ? $"{elapsed.TotalSeconds:F1}s" : $"{(int)elapsed.TotalMinutes}m {elapsed.Seconds}s"; + + private void EnsureAssistantSegment() + { + if (_assistantSegmentId.Value != 0) + return; + + _assistantBuffer.Clear(); + _assistantSegmentId = NextSegmentId(); + _chatHistory.AppendTracked(_assistantSegmentId, + new StaticTextSegment("Netclaw: ", Color.White)); + } + + private void FinalizeAssistantSegmentIfNeeded() + { + if (_assistantSegmentId.Value == 0) + return; + + _chatHistory.Replace(_assistantSegmentId, + new StaticTextSegment($"Netclaw: {_assistantBuffer}", Color.White), + keepTracked: false); + _assistantSegmentId = default; + _assistantBuffer.Clear(); + _chatHistory.AppendLine(""); + } } diff --git a/src/Netclaw.Cli/Tui/ChatViewModel.cs b/src/Netclaw.Cli/Tui/ChatViewModel.cs index eb627fcf2..97e28e761 100644 --- a/src/Netclaw.Cli/Tui/ChatViewModel.cs +++ b/src/Netclaw.Cli/Tui/ChatViewModel.cs @@ -1,31 +1,27 @@ using System.Reactive.Linq; using System.Reactive.Subjects; -using Akka.Actor; -using Akka.Streams; -using Akka.Streams.Dsl; using Microsoft.Extensions.AI; using Netclaw.Actors.Channels; using Netclaw.Actors.Protocol; +using Netclaw.Cli.Daemon; using Netclaw.Configuration; using Termina.Reactive; namespace Netclaw.Cli.Tui; /// -/// Reactive ViewModel for the chat page. Uses -/// directly (in-process, no SignalR indirection). Manages session lifecycle, -/// input submission, and output forwarding to the page. +/// Reactive ViewModel for the chat page. Uses +/// to talk to the daemon-hosted session hub over SignalR. /// public partial class ChatViewModel : ReactiveViewModel { - private readonly SessionPipeline _pipeline; - private readonly ActorSystem _system; + private readonly DaemonClient _daemonClient; private readonly TimeProvider _timeProvider; private readonly SessionConfig _sessionConfig; - private MaterializedSession? _session; - private ISourceQueueWithComplete? _inputQueue; private readonly Subject _outputSubject = new(); + private IDisposable? _daemonOutputSubscription; + private IDisposable? _daemonConnectionSubscription; #pragma warning disable CS0169, CS0414 // Backing fields used by [Reactive] source generator [Reactive] private bool _isGenerating; @@ -48,13 +44,11 @@ public partial class ChatViewModel : ReactiveViewModel public int ContextWindowTokens => _sessionConfig.ContextWindowTokens; public ChatViewModel( - SessionPipeline pipeline, - ActorSystem system, + DaemonClient daemonClient, TimeProvider timeProvider, SessionConfig sessionConfig) { - _pipeline = pipeline; - _system = system; + _daemonClient = daemonClient; _timeProvider = timeProvider; _sessionConfig = sessionConfig; } @@ -69,17 +63,8 @@ private async Task InitializeSessionAsync() { try { - var sessionId = new SessionId($"tui/{Guid.NewGuid():N}"); - SessionIdDisplay = sessionId.Value; - - _session = await _pipeline.CreateAsync(sessionId, new SessionPipelineOptions - { - ChannelType = "tui" - }); - - // Materialize output stream → forward to Subject for page rendering - _session.Output - .To(Sink.ForEach(output => + _daemonOutputSubscription = _daemonClient.SessionOutput + .Subscribe(output => { _outputSubject.OnNext(output); @@ -95,13 +80,26 @@ private async Task InitializeSessionAsync() } RequestRedraw(); - })) - .Run(_system); + }); - // Materialize input with queue for imperative push - _inputQueue = Source.Queue(16, OverflowStrategy.Backpressure) - .ToMaterialized(_session.Input, Keep.Left) - .Run(_system); + _daemonConnectionSubscription = _daemonClient.ConnectionEvents + .Subscribe(evt => + { + if (IsGenerating && evt.State is DaemonConnectionState.Connected) + { + StatusMessage = "Generating..."; + } + else + { + StatusMessage = evt.Message; + } + + RequestRedraw(); + }); + + await ConnectWithRetryAsync(); + var sessionId = await _daemonClient.CreateSessionAsync("tui"); + SessionIdDisplay = sessionId; StatusMessage = "Ready"; RequestRedraw(); @@ -118,18 +116,29 @@ private async Task InitializeSessionAsync() /// public async Task SubmitAsync(string text) { - if (_inputQueue is null || string.IsNullOrWhiteSpace(text)) + if (string.IsNullOrWhiteSpace(text)) return; IsGenerating = true; StatusMessage = "Generating..."; - await _inputQueue.OfferAsync(new ChannelInput + try + { + await _daemonClient.EnsureSessionAsync("tui"); + + await _daemonClient.SendAsync(new ChannelInput + { + SenderId = "local-user", + Contents = [new TextContent(text)], + ReceivedAt = _timeProvider.GetUtcNow() + }); + } + catch (Exception ex) { - SenderId = "local-user", - Contents = [new TextContent(text)], - ReceivedAt = _timeProvider.GetUtcNow() - }); + IsGenerating = false; + StatusMessage = $"Connection failed: {ex.Message}"; + RequestRedraw(); + } } public void RequestAppShutdown() @@ -139,13 +148,39 @@ public void RequestAppShutdown() public override void Dispose() { + _daemonOutputSubscription?.Dispose(); + _daemonConnectionSubscription?.Dispose(); _outputSubject.Dispose(); - if (_session is not null) - { - _ = _session.DisposeAsync(); - } DisposeReactiveFields(); base.Dispose(); } + + private async Task ConnectWithRetryAsync() + { + var delays = new[] + { + TimeSpan.FromSeconds(1), + TimeSpan.FromSeconds(2), + TimeSpan.FromSeconds(5), + TimeSpan.FromSeconds(10) + }; + + for (var attempt = 0; attempt <= delays.Length; attempt++) + { + try + { + await _daemonClient.ConnectAsync(); + return; + } + catch when (attempt < delays.Length) + { + StatusMessage = $"Connecting... retry {attempt + 1}/{delays.Length}"; + RequestRedraw(); + await Task.Delay(delays[attempt]); + } + } + + throw new InvalidOperationException("Unable to connect to daemon after retries."); + } } diff --git a/src/Netclaw.Daemon/Gateway/SessionConnectionMap.cs b/src/Netclaw.Daemon/Gateway/SessionConnectionMap.cs index 0c5db2bca..df28117d0 100644 --- a/src/Netclaw.Daemon/Gateway/SessionConnectionMap.cs +++ b/src/Netclaw.Daemon/Gateway/SessionConnectionMap.cs @@ -81,6 +81,15 @@ public void Disconnect(SignalRConnectionId connectionId) RemoveConnectionInternal(connectionId); } + public void Clear() + { + lock (_gate) + { + _sessionToConnection.Clear(); + _connectionToSession.Clear(); + } + } + private void RemoveConnectionInternal(SignalRConnectionId connectionId) { if (_connectionToSession.TryGetValue(connectionId, out var sessionId)) diff --git a/src/Netclaw.Daemon/Gateway/SessionHub.cs b/src/Netclaw.Daemon/Gateway/SessionHub.cs index 86c96aa24..c11057aa1 100644 --- a/src/Netclaw.Daemon/Gateway/SessionHub.cs +++ b/src/Netclaw.Daemon/Gateway/SessionHub.cs @@ -1,4 +1,5 @@ using Microsoft.AspNetCore.SignalR; +using Netclaw.Actors.Protocol; namespace Netclaw.Daemon.Gateway; @@ -35,6 +36,11 @@ public Task CreateSession(string channelType) return _registry.CreateSessionAsync(Context.ConnectionId, channelType); } + public Task EnsureSession(string? sessionId, string channelType) + { + return _registry.EnsureSessionAsync(Context.ConnectionId, sessionId, channelType); + } + public Task AttachSession(string sessionId) { return _registry.AttachSessionAsync(Context.ConnectionId, sessionId); diff --git a/src/Netclaw.Daemon/Gateway/SessionOutputMapper.cs b/src/Netclaw.Daemon/Gateway/SessionOutputMapper.cs index 178e43b18..8a07f9301 100644 --- a/src/Netclaw.Daemon/Gateway/SessionOutputMapper.cs +++ b/src/Netclaw.Daemon/Gateway/SessionOutputMapper.cs @@ -18,6 +18,14 @@ public static class SessionOutputMapper Text = msg.Text }, + TextDeltaOutput msg => new SessionOutputDto + { + Type = "text_delta", + SessionId = msg.SessionId.Value, + TimestampMs = msg.TimestampMs, + Text = msg.Delta + }, + ThinkingOutput msg => new SessionOutputDto { Type = "thinking", @@ -26,6 +34,14 @@ public static class SessionOutputMapper Text = msg.Text }, + ThinkingDeltaOutput msg => new SessionOutputDto + { + Type = "thinking_delta", + SessionId = msg.SessionId.Value, + TimestampMs = msg.TimestampMs, + Text = msg.Delta + }, + ToolCallOutput msg => new SessionOutputDto { Type = "tool_call", diff --git a/src/Netclaw.Daemon/Gateway/SessionRegistry.cs b/src/Netclaw.Daemon/Gateway/SessionRegistry.cs index c79a50827..b1cac6178 100644 --- a/src/Netclaw.Daemon/Gateway/SessionRegistry.cs +++ b/src/Netclaw.Daemon/Gateway/SessionRegistry.cs @@ -82,6 +82,35 @@ public async Task CreateSessionAsync(string connectionId, string channel return sessionId.Value; } + public async Task EnsureSessionAsync( + string connectionId, + string? sessionId, + string channelType) + { + var callerConnectionId = ParseConnectionId(connectionId); + + if (!string.IsNullOrWhiteSpace(sessionId)) + { + var requestedSessionId = ParseSessionId(sessionId); + if (_sessions.ContainsKey(requestedSessionId)) + { + _connections.AttachSession(requestedSessionId, callerConnectionId); + return new SessionEnsureResultDto + { + SessionId = requestedSessionId.Value, + Created = false + }; + } + } + + var createdSessionId = await CreateSessionAsync(connectionId, channelType); + return new SessionEnsureResultDto + { + SessionId = createdSessionId, + Created = true + }; + } + /// /// Attaches the current SignalR connection to an existing session. /// Supports reconnect flows where connection IDs rotate. @@ -140,6 +169,29 @@ public Task OnDisconnectedAsync(string connectionId) return Task.CompletedTask; } + public async Task ShutdownAsync(CancellationToken cancellationToken) + { + var sessions = _sessions.ToArray(); + _sessions.Clear(); + _connections.Clear(); + + var disposeTasks = sessions + .Select(x => x.Value.Session.DisposeAsync().AsTask()) + .ToArray(); + + if (disposeTasks.Length == 0) + return; + + try + { + await Task.WhenAll(disposeTasks).WaitAsync(cancellationToken); + } + catch (OperationCanceledException) + { + _logger.LogWarning("Timed out while disposing {Count} active session(s) during shutdown.", disposeTasks.Length); + } + } + private void PublishOutput(SessionId sessionId, SessionOutput output) { if (!_sessions.ContainsKey(sessionId)) diff --git a/src/Netclaw.Daemon/Program.cs b/src/Netclaw.Daemon/Program.cs index 4ef83a3a9..e069ac692 100644 --- a/src/Netclaw.Daemon/Program.cs +++ b/src/Netclaw.Daemon/Program.cs @@ -118,6 +118,11 @@ static NetclawPaths ConfigureConfigServices(IServiceCollection services, IConfig static void ConfigureDaemonServices(IServiceCollection services, IConfigurationManager configuration, NetclawPaths paths) { + services.Configure(options => + { + options.ShutdownTimeout = TimeSpan.FromSeconds(10); + }); + // Resolve models for session config var models = configuration.GetSection("Models") .Get() ?? new ModelSelection(); @@ -178,4 +183,8 @@ static void ConfigureDaemonServices(IServiceCollection services, IConfigurationM // PID file authority for daemon lifecycle management services.AddSingleton(); services.AddSingleton(sp => sp.GetRequiredService()); + + // Active session cleanup during host shutdown + services.AddSingleton(); + services.AddSingleton(sp => sp.GetRequiredService()); } diff --git a/src/Netclaw.Daemon/Services/SessionRegistryShutdownService.cs b/src/Netclaw.Daemon/Services/SessionRegistryShutdownService.cs new file mode 100644 index 000000000..9f60194bb --- /dev/null +++ b/src/Netclaw.Daemon/Services/SessionRegistryShutdownService.cs @@ -0,0 +1,40 @@ +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Netclaw.Daemon.Gateway; + +namespace Netclaw.Daemon.Services; + +/// +/// Ensures active SignalR sessions are disposed promptly during host shutdown. +/// +public sealed class SessionRegistryShutdownService : IHostedService +{ + private readonly SessionRegistry _registry; + private readonly ILogger _logger; + + public SessionRegistryShutdownService( + SessionRegistry registry, + ILogger logger) + { + _registry = registry; + _logger = logger; + } + + public Task StartAsync(CancellationToken cancellationToken) + => Task.CompletedTask; + + public async Task StopAsync(CancellationToken cancellationToken) + { + using var hardCutoff = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + hardCutoff.CancelAfter(TimeSpan.FromSeconds(8)); + + try + { + await _registry.ShutdownAsync(hardCutoff.Token); + } + catch (OperationCanceledException) + { + _logger.LogWarning("Session shutdown hit hard cutoff during daemon stop."); + } + } +}