From 4b3c050f0e7b77733fb40959a65fa6229d5e5f10 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Tue, 24 Feb 2026 13:46:28 -0600 Subject: [PATCH 1/4] Switch CLI chat path to daemon SignalR client Replace in-process session pipeline usage in chat and headless modes with a daemon-backed SignalR client, including reconnect handling and DTO mapping. Remove transitional CLI provider/client factory shims, add SignalR client dependency, and add mapping tests for daemon output types. --- Directory.Packages.props | 3 +- .../Cli/DaemonClientMappingTests.cs | 47 ++++ .../Configuration/ChatClientFactory.cs | 33 --- .../NetclawChatClientProvider.cs | 31 --- src/Netclaw.Cli/Daemon/DaemonClient.cs | 230 ++++++++++++++++++ src/Netclaw.Cli/HeadlessChannel.cs | 75 +++--- src/Netclaw.Cli/Netclaw.Cli.csproj | 8 +- src/Netclaw.Cli/Program.cs | 72 +----- src/Netclaw.Cli/Tui/ChatViewModel.cs | 98 ++++---- 9 files changed, 391 insertions(+), 206 deletions(-) create mode 100644 src/Netclaw.Actors.Tests/Cli/DaemonClientMappingTests.cs delete mode 100644 src/Netclaw.Cli/Configuration/ChatClientFactory.cs delete mode 100644 src/Netclaw.Cli/Configuration/NetclawChatClientProvider.cs create mode 100644 src/Netclaw.Cli/Daemon/DaemonClient.cs diff --git a/Directory.Packages.props b/Directory.Packages.props index e7ac10898..fa8516d4c 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -13,6 +13,7 @@ + @@ -37,4 +38,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..693522296 --- /dev/null +++ b/src/Netclaw.Actors.Tests/Cli/DaemonClientMappingTests.cs @@ -0,0 +1,47 @@ +using Netclaw.Actors.Protocol; +using Netclaw.Cli.Daemon; +using Xunit; + +namespace Netclaw.Actors.Tests.Cli; + +public sealed class DaemonClientMappingTests +{ + [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.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..c5c1f5e82 --- /dev/null +++ b/src/Netclaw.Cli/Daemon/DaemonClient.cs @@ -0,0 +1,230 @@ +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 Subject _outputSubject = new(); + private readonly SemaphoreSlim _connectGate = new(1, 1); + + private string? _sessionId; + + public DaemonClient(string daemonEndpoint) + { + if (string.IsNullOrWhiteSpace(daemonEndpoint)) + throw new ArgumentException("Daemon endpoint cannot be empty.", nameof(daemonEndpoint)); + + var hubUrl = BuildHubUrl(daemonEndpoint); + + _connection = new HubConnectionBuilder() + .WithUrl(hubUrl) + .WithAutomaticReconnect(ReconnectDelays) + .Build(); + + _connection.On("ReceiveOutput", dto => + { + _outputSubject.OnNext(FromDto(dto)); + }); + + _connection.Reconnected += async _ => + { + var sessionId = _sessionId; + if (!string.IsNullOrWhiteSpace(sessionId)) + await _connection.InvokeCoreAsync("AttachSession", [sessionId]); + }; + } + + public IObservable SessionOutput => _outputSubject.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; + + Exception? lastError = null; + foreach (var delay in ReconnectDelays) + { + if (delay > TimeSpan.Zero) + await Task.Delay(delay, cancellationToken); + + try + { + await _connection.StartAsync(cancellationToken); + return; + } + catch (Exception ex) + { + lastError = ex; + } + } + + throw new InvalidOperationException("Failed to connect to daemon SignalR hub.", lastError); + } + finally + { + _connectGate.Release(); + } + } + + public async Task CreateSessionAsync( + string channelType, + CancellationToken cancellationToken = default) + { + if (string.IsNullOrWhiteSpace(channelType)) + throw new ArgumentException("Channel type cannot be empty.", nameof(channelType)); + + await ConnectAsync(cancellationToken); + var sessionId = await _connection.InvokeCoreAsync( + "CreateSession", + [channelType], + cancellationToken); + + _sessionId = sessionId; + return sessionId; + } + + 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(); + await _connection.DisposeAsync(); + _connectGate.Dispose(); + } + + private static string BuildHubUrl(string endpoint) + { + var trimmed = endpoint.TrimEnd('/'); + return $"{trimmed}/hub/session"; + } + + 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 + }, + "thinking" => new ThinkingOutput + { + SessionId = sessionId, + TimestampMs = dto.TimestampMs, + Text = 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/HeadlessChannel.cs b/src/Netclaw.Cli/HeadlessChannel.cs index ac8a77fda..f597553f5 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,31 @@ 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; 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 +55,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 +69,48 @@ 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 subscription = _daemonClient.SessionOutput.Subscribe(output => { - ChannelType = ChannelType - }, stopping); + HandleOutput(output, logWriter); + if (output is TurnCompleted) + turnCompleted.TrySetResult(); + }); - // Materialize output stream → console + disk logging, exit on TurnCompleted - _session.Output - .To(Sink.ForEach(output => HandleOutput(output, logWriter))) - .Run(_system); + 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(); } } @@ -159,7 +158,6 @@ private void HandleOutput(SessionOutput output, StreamWriter log) case TurnCompleted msg: Log(log, $"TURN_COMPLETED: turn={msg.TurnNumber}"); Log(log, "SESSION_ENDED"); - _lifetime.StopApplication(); break; case CompactionOutput msg: @@ -173,4 +171,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 + { + // Ignore secondary logging failures. + } + } } 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/ChatViewModel.cs b/src/Netclaw.Cli/Tui/ChatViewModel.cs index eb627fcf2..c711c43d9 100644 --- a/src/Netclaw.Cli/Tui/ChatViewModel.cs +++ b/src/Netclaw.Cli/Tui/ChatViewModel.cs @@ -1,31 +1,26 @@ 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; #pragma warning disable CS0169, CS0414 // Backing fields used by [Reactive] source generator [Reactive] private bool _isGenerating; @@ -48,13 +43,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 +62,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 +79,11 @@ 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); + await ConnectWithRetryAsync(); + var sessionId = await _daemonClient.CreateSessionAsync("tui"); + SessionIdDisplay = sessionId; StatusMessage = "Ready"; RequestRedraw(); @@ -118,18 +100,27 @@ 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 { - SenderId = "local-user", - Contents = [new TextContent(text)], - ReceivedAt = _timeProvider.GetUtcNow() - }); + await _daemonClient.SendAsync(new ChannelInput + { + SenderId = "local-user", + Contents = [new TextContent(text)], + ReceivedAt = _timeProvider.GetUtcNow() + }); + } + catch (Exception ex) + { + IsGenerating = false; + StatusMessage = $"Connection failed: {ex.Message}"; + RequestRedraw(); + } } public void RequestAppShutdown() @@ -139,13 +130,38 @@ public void RequestAppShutdown() public override void Dispose() { + _daemonOutputSubscription?.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."); + } } From cac8accb2e656971e84729d068ab3e8f85956374 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Tue, 24 Feb 2026 14:08:08 -0600 Subject: [PATCH 2/4] Surface daemon websocket reconnect status in CLI Publish daemon connection lifecycle events from the SignalR client and wire them into TUI/headless status reporting so users see connect, reconnect, and disconnect states. Add automatic reconnect loop after socket closure and log connection transitions in headless session output. --- src/Netclaw.Cli/Daemon/DaemonClient.cs | 78 +++++++++++++++++++ .../Daemon/DaemonConnectionEvent.cs | 11 +++ src/Netclaw.Cli/HeadlessChannel.cs | 9 ++- src/Netclaw.Cli/Tui/ChatPage.cs | 6 +- src/Netclaw.Cli/Tui/ChatViewModel.cs | 17 ++++ 5 files changed, 118 insertions(+), 3 deletions(-) create mode 100644 src/Netclaw.Cli/Daemon/DaemonConnectionEvent.cs diff --git a/src/Netclaw.Cli/Daemon/DaemonClient.cs b/src/Netclaw.Cli/Daemon/DaemonClient.cs index c5c1f5e82..b47006761 100644 --- a/src/Netclaw.Cli/Daemon/DaemonClient.cs +++ b/src/Netclaw.Cli/Daemon/DaemonClient.cs @@ -24,9 +24,13 @@ public sealed class DaemonClient : IAsyncDisposable private readonly HubConnection _connection; 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 bool _hasConnected; + private bool _disposed; public DaemonClient(string daemonEndpoint) { @@ -50,10 +54,38 @@ public DaemonClient(string daemonEndpoint) var sessionId = _sessionId; if (!string.IsNullOrWhiteSpace(sessionId)) await _connection.InvokeCoreAsync("AttachSession", [sessionId]); + + _connectionSubject.OnNext(new DaemonConnectionEvent( + DaemonConnectionState.Connected, + "Reconnected to daemon.")); + }; + + _connection.Reconnecting += ex => + { + var reason = ex?.Message ?? "connection dropped"; + _connectionSubject.OnNext(new DaemonConnectionEvent( + DaemonConnectionState.Reconnecting, + $"Reconnecting to daemon: {reason}")); + return Task.CompletedTask; + }; + + _connection.Closed += async ex => + { + if (_disposed) + return; + + var reason = ex?.Message ?? "connection closed"; + _connectionSubject.OnNext(new DaemonConnectionEvent( + DaemonConnectionState.Disconnected, + $"Disconnected from daemon: {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; @@ -68,6 +100,10 @@ public async Task ConnectAsync(CancellationToken cancellationToken = default) if (IsConnected) return; + _connectionSubject.OnNext(new DaemonConnectionEvent( + DaemonConnectionState.Connecting, + "Connecting to daemon...")); + Exception? lastError = null; foreach (var delay in ReconnectDelays) { @@ -77,6 +113,15 @@ public async Task ConnectAsync(CancellationToken cancellationToken = default) try { await _connection.StartAsync(cancellationToken); + + var sessionId = _sessionId; + if (!string.IsNullOrWhiteSpace(sessionId)) + await _connection.InvokeCoreAsync("AttachSession", [sessionId], cancellationToken); + + _connectionSubject.OnNext(new DaemonConnectionEvent( + DaemonConnectionState.Connected, + _hasConnected ? "Reconnected to daemon." : "Connected to daemon.")); + _hasConnected = true; return; } catch (Exception ex) @@ -133,10 +178,43 @@ await _connection.InvokeCoreAsync( 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; + + var attempts = 0; + while (!_disposed && !_lifetimeCts.Token.IsCancellationRequested) + { + attempts++; + try + { + await ConnectAsync(_lifetimeCts.Token); + return; + } + catch when (!_disposed && !_lifetimeCts.Token.IsCancellationRequested) + { + if (attempts >= 20) + { + _connectionSubject.OnNext(new DaemonConnectionEvent( + DaemonConnectionState.Disconnected, + "Unable to reconnect to daemon after multiple attempts.")); + return; + } + + await Task.Delay(TimeSpan.FromSeconds(2), _lifetimeCts.Token); + } + } + } + private static string BuildHubUrl(string endpoint) { var trimmed = endpoint.TrimEnd('/'); diff --git a/src/Netclaw.Cli/Daemon/DaemonConnectionEvent.cs b/src/Netclaw.Cli/Daemon/DaemonConnectionEvent.cs new file mode 100644 index 000000000..154b4caf3 --- /dev/null +++ b/src/Netclaw.Cli/Daemon/DaemonConnectionEvent.cs @@ -0,0 +1,11 @@ +namespace Netclaw.Cli.Daemon; + +public enum DaemonConnectionState +{ + Connecting, + Connected, + Reconnecting, + Disconnected +} + +public sealed record DaemonConnectionEvent(DaemonConnectionState State, string Message); diff --git a/src/Netclaw.Cli/HeadlessChannel.cs b/src/Netclaw.Cli/HeadlessChannel.cs index f597553f5..d0a11dd20 100644 --- a/src/Netclaw.Cli/HeadlessChannel.cs +++ b/src/Netclaw.Cli/HeadlessChannel.cs @@ -76,6 +76,11 @@ private async Task RunHeadlessAsync(CancellationToken stopping) var turnCompleted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var connectionSubscription = _daemonClient.ConnectionEvents.Subscribe(evt => + { + Log(logWriter, $"CONNECTION: {evt.Message}"); + }); + using var subscription = _daemonClient.SessionOutput.Subscribe(output => { HandleOutput(output, logWriter); @@ -181,9 +186,9 @@ private void WriteFailureLog(string kind, Exception ex) File.AppendAllText(path, $"[{_timeProvider.GetUtcNow():o}] {kind}: {ex}\n"); } - catch + catch (Exception logEx) { - // Ignore secondary logging failures. + Console.Error.WriteLine($"[headless:error] Failed to write failure log: {logEx.Message}"); } } } diff --git a/src/Netclaw.Cli/Tui/ChatPage.cs b/src/Netclaw.Cli/Tui/ChatPage.cs index 46115af20..e1754075f 100644 --- a/src/Netclaw.Cli/Tui/ChatPage.cs +++ b/src/Netclaw.Cli/Tui/ChatPage.cs @@ -113,7 +113,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 diff --git a/src/Netclaw.Cli/Tui/ChatViewModel.cs b/src/Netclaw.Cli/Tui/ChatViewModel.cs index c711c43d9..3655b2c02 100644 --- a/src/Netclaw.Cli/Tui/ChatViewModel.cs +++ b/src/Netclaw.Cli/Tui/ChatViewModel.cs @@ -21,6 +21,7 @@ public partial class ChatViewModel : ReactiveViewModel 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; @@ -81,6 +82,21 @@ private async Task InitializeSessionAsync() RequestRedraw(); }); + _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; @@ -131,6 +147,7 @@ public void RequestAppShutdown() public override void Dispose() { _daemonOutputSubscription?.Dispose(); + _daemonConnectionSubscription?.Dispose(); _outputSubject.Dispose(); DisposeReactiveFields(); From 2cc9b80a1b8fbba7b41868ff63970bd9927f6e20 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Tue, 24 Feb 2026 15:02:10 -0600 Subject: [PATCH 3/4] Add typed session recovery and streamed turn output Introduce typed SignalR session ensure semantics between CLI and daemon, add reconnect integration coverage, and stream assistant deltas while preserving turn completion boundaries. Update TUI/headless rendering to consume delta outputs with compatibility for final snapshots and keep reconnect status visible to users. --- Directory.Packages.props | 1 + .../Cli/DaemonClientMappingTests.cs | 18 ++ .../DaemonClientReconnectIntegrationTests.cs | 175 ++++++++++++++++++ .../Netclaw.Actors.Tests.csproj | 1 + .../Sessions/LlmSessionIntegrationTests.cs | 15 +- .../Protocol/SessionEnsureResultDto.cs | 14 ++ src/Netclaw.Actors/Protocol/SessionOutput.cs | 18 ++ .../Protocol/SessionOutputDto.cs | 2 +- src/Netclaw.Actors/Sessions/LlmMessages.cs | 12 ++ .../Sessions/LlmSessionActor.cs | 144 +++++++++++++- src/Netclaw.Cli/Daemon/DaemonClient.cs | 137 +++++++++++--- .../Daemon/DaemonConnectionEvent.cs | 8 +- src/Netclaw.Cli/HeadlessChannel.cs | 29 +++ src/Netclaw.Cli/Tui/ChatPage.cs | 51 ++++- src/Netclaw.Cli/Tui/ChatViewModel.cs | 2 + src/Netclaw.Daemon/Gateway/SessionHub.cs | 6 + .../Gateway/SessionOutputMapper.cs | 16 ++ src/Netclaw.Daemon/Gateway/SessionRegistry.cs | 29 +++ 18 files changed, 642 insertions(+), 36 deletions(-) create mode 100644 src/Netclaw.Actors.Tests/Cli/DaemonClientReconnectIntegrationTests.cs create mode 100644 src/Netclaw.Actors/Protocol/SessionEnsureResultDto.cs diff --git a/Directory.Packages.props b/Directory.Packages.props index fa8516d4c..5253d4491 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -14,6 +14,7 @@ + diff --git a/src/Netclaw.Actors.Tests/Cli/DaemonClientMappingTests.cs b/src/Netclaw.Actors.Tests/Cli/DaemonClientMappingTests.cs index 693522296..cabc61b90 100644 --- a/src/Netclaw.Actors.Tests/Cli/DaemonClientMappingTests.cs +++ b/src/Netclaw.Actors.Tests/Cli/DaemonClientMappingTests.cs @@ -6,6 +6,24 @@ 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() { 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/Daemon/DaemonClient.cs b/src/Netclaw.Cli/Daemon/DaemonClient.cs index b47006761..40f3d2927 100644 --- a/src/Netclaw.Cli/Daemon/DaemonClient.cs +++ b/src/Netclaw.Cli/Daemon/DaemonClient.cs @@ -23,12 +23,15 @@ public sealed class DaemonClient : IAsyncDisposable ]; 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; @@ -37,10 +40,11 @@ public DaemonClient(string daemonEndpoint) if (string.IsNullOrWhiteSpace(daemonEndpoint)) throw new ArgumentException("Daemon endpoint cannot be empty.", nameof(daemonEndpoint)); - var hubUrl = BuildHubUrl(daemonEndpoint); + _daemonEndpoint = daemonEndpoint.TrimEnd('/'); + _hubUrl = BuildHubUrl(_daemonEndpoint); _connection = new HubConnectionBuilder() - .WithUrl(hubUrl) + .WithUrl(_hubUrl) .WithAutomaticReconnect(ReconnectDelays) .Build(); @@ -51,13 +55,13 @@ public DaemonClient(string daemonEndpoint) _connection.Reconnected += async _ => { - var sessionId = _sessionId; - if (!string.IsNullOrWhiteSpace(sessionId)) - await _connection.InvokeCoreAsync("AttachSession", [sessionId]); + if (!string.IsNullOrWhiteSpace(_channelType)) + await EnsureSessionInternalAsync(_channelType!, CancellationToken.None); _connectionSubject.OnNext(new DaemonConnectionEvent( DaemonConnectionState.Connected, - "Reconnected to daemon.")); + _daemonEndpoint, + $"Reconnected to daemon at {_daemonEndpoint}.")); }; _connection.Reconnecting += ex => @@ -65,7 +69,8 @@ public DaemonClient(string daemonEndpoint) var reason = ex?.Message ?? "connection dropped"; _connectionSubject.OnNext(new DaemonConnectionEvent( DaemonConnectionState.Reconnecting, - $"Reconnecting to daemon: {reason}")); + _daemonEndpoint, + $"Reconnecting to {_daemonEndpoint}: {reason}")); return Task.CompletedTask; }; @@ -77,7 +82,8 @@ public DaemonClient(string daemonEndpoint) var reason = ex?.Message ?? "connection closed"; _connectionSubject.OnNext(new DaemonConnectionEvent( DaemonConnectionState.Disconnected, - $"Disconnected from daemon: {reason}")); + _daemonEndpoint, + $"Disconnected from daemon at {_daemonEndpoint}: {reason}")); if (!string.IsNullOrWhiteSpace(_sessionId)) await ReconnectLoopAsync(); @@ -100,9 +106,18 @@ public async Task ConnectAsync(CancellationToken cancellationToken = default) 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, - "Connecting to daemon...")); + _daemonEndpoint, + $"Connecting to daemon at {_daemonEndpoint}...")); Exception? lastError = null; foreach (var delay in ReconnectDelays) @@ -114,13 +129,12 @@ public async Task ConnectAsync(CancellationToken cancellationToken = default) { await _connection.StartAsync(cancellationToken); - var sessionId = _sessionId; - if (!string.IsNullOrWhiteSpace(sessionId)) - await _connection.InvokeCoreAsync("AttachSession", [sessionId], cancellationToken); - _connectionSubject.OnNext(new DaemonConnectionEvent( DaemonConnectionState.Connected, - _hasConnected ? "Reconnected to daemon." : "Connected to daemon.")); + _daemonEndpoint, + _hasConnected + ? $"Reconnected to daemon at {_daemonEndpoint}." + : $"Connected to daemon at {_daemonEndpoint}.")); _hasConnected = true; return; } @@ -138,6 +152,18 @@ public async Task ConnectAsync(CancellationToken cancellationToken = default) } } + 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) @@ -145,14 +171,17 @@ public async Task CreateSessionAsync( if (string.IsNullOrWhiteSpace(channelType)) throw new ArgumentException("Channel type cannot be empty.", nameof(channelType)); - await ConnectAsync(cancellationToken); - var sessionId = await _connection.InvokeCoreAsync( - "CreateSession", - [channelType], - cancellationToken); + _channelType = channelType; + _sessionId = null; + return await EnsureSessionInternalAsync(channelType, cancellationToken); + } - _sessionId = sessionId; - return sessionId; + 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) @@ -191,26 +220,50 @@ 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 >= 20) + if (attempts >= maxAttempts) { _connectionSubject.OnNext(new DaemonConnectionEvent( DaemonConnectionState.Disconnected, - "Unable to reconnect to daemon after multiple attempts.")); + _daemonEndpoint, + $"Unable to reconnect to daemon at {_daemonEndpoint} after {maxAttempts} attempts.", + attempts, + maxAttempts, + 0)); return; } - await Task.Delay(TimeSpan.FromSeconds(2), _lifetimeCts.Token); + 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); + } } } } @@ -221,6 +274,30 @@ private static string BuildHubUrl(string endpoint) 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); @@ -233,12 +310,24 @@ internal static SessionOutput FromDto(SessionOutputDto dto) 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, diff --git a/src/Netclaw.Cli/Daemon/DaemonConnectionEvent.cs b/src/Netclaw.Cli/Daemon/DaemonConnectionEvent.cs index 154b4caf3..35a99e331 100644 --- a/src/Netclaw.Cli/Daemon/DaemonConnectionEvent.cs +++ b/src/Netclaw.Cli/Daemon/DaemonConnectionEvent.cs @@ -8,4 +8,10 @@ public enum DaemonConnectionState Disconnected } -public sealed record DaemonConnectionEvent(DaemonConnectionState State, string Message); +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/HeadlessChannel.cs b/src/Netclaw.Cli/HeadlessChannel.cs index d0a11dd20..ba3ecba71 100644 --- a/src/Netclaw.Cli/HeadlessChannel.cs +++ b/src/Netclaw.Cli/HeadlessChannel.cs @@ -23,6 +23,8 @@ public sealed class HeadlessChannel : IChannel private readonly ILogger _logger; private bool _isConnected; + private bool _receivedTextDeltaInCurrentTurn; + private bool _receivedThinkingDeltaInCurrentTurn; public string ChannelType => "headless"; public string DisplayName => "Headless Prompt"; @@ -129,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 ?? "{}"}"); @@ -161,8 +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"); + _receivedTextDeltaInCurrentTurn = false; + _receivedThinkingDeltaInCurrentTurn = false; break; case CompactionOutput msg: diff --git a/src/Netclaw.Cli/Tui/ChatPage.cs b/src/Netclaw.Cli/Tui/ChatPage.cs index e1754075f..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(); @@ -177,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. @@ -241,6 +265,7 @@ private void HandleOutput(SessionOutput output) case TurnCompleted: RemoveThinkingSpinner(); + FinalizeAssistantSegmentIfNeeded(); ViewModel.StatusMessage = "Ready"; _chatHistory.ScrollToBottom(); break; @@ -274,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 3655b2c02..97e28e761 100644 --- a/src/Netclaw.Cli/Tui/ChatViewModel.cs +++ b/src/Netclaw.Cli/Tui/ChatViewModel.cs @@ -124,6 +124,8 @@ public async Task SubmitAsync(string text) try { + await _daemonClient.EnsureSessionAsync("tui"); + await _daemonClient.SendAsync(new ChannelInput { SenderId = "local-user", 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..717e99741 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. From 1b830299e419e3ae16646144b0e28c0675e7d98a Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Tue, 24 Feb 2026 15:08:51 -0600 Subject: [PATCH 4/4] Enforce daemon shutdown hard cutoff with session cleanup Add bounded session cleanup during host shutdown and configure a 10-second daemon shutdown timeout. Escalate daemon stop from graceful SIGTERM to hard-kill cutoff so stop commands do not hang behind long-lived chat connections. --- src/Netclaw.Cli/Daemon/DaemonManager.cs | 20 ++++++++-- .../Gateway/SessionConnectionMap.cs | 9 +++++ src/Netclaw.Daemon/Gateway/SessionRegistry.cs | 23 +++++++++++ src/Netclaw.Daemon/Program.cs | 9 +++++ .../SessionRegistryShutdownService.cs | 40 +++++++++++++++++++ 5 files changed, 97 insertions(+), 4 deletions(-) create mode 100644 src/Netclaw.Daemon/Services/SessionRegistryShutdownService.cs 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.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/SessionRegistry.cs b/src/Netclaw.Daemon/Gateway/SessionRegistry.cs index 717e99741..b1cac6178 100644 --- a/src/Netclaw.Daemon/Gateway/SessionRegistry.cs +++ b/src/Netclaw.Daemon/Gateway/SessionRegistry.cs @@ -169,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."); + } + } +}