Complete CLI daemon SignalR adapter with streaming and typed session recovery

Directory.Packages.props

@@ -13,6 +13,8 @@
     <PackageVersion Include="Akka.Persistence.Hosting" Version="$(AkkaHostingVersion)" />
     <PackageVersion Include="Aaron.Akka.Reminders" Version="$(AkkaRemindersVersion)" />
     <PackageVersion Include="Microsoft.Extensions.AI.Abstractions" Version="10.3.0" />
+    <PackageVersion Include="Microsoft.AspNetCore.SignalR.Client" Version="10.0.2" />
+    <PackageVersion Include="Microsoft.AspNetCore.TestHost" Version="10.0.2" />
     <PackageVersion Include="Microsoft.Extensions.Configuration.Json" Version="10.0.2" />
     <PackageVersion Include="Microsoft.Extensions.Hosting" Version="10.0.2" />
     <PackageVersion Include="OllamaSharp" Version="5.4.16" />
@@ -37,4 +39,4 @@
   <ItemGroup>
     <PackageVersion Include="Microsoft.SourceLink.GitHub" Version="10.0.103" />
   </ItemGroup>
-</Project>
+</Project>

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<TextDeltaOutput>(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<ToolResultOutput>(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<ErrorOutput>(output);
+        Assert.Contains("Unknown output type", error.Message);
+        Assert.Equal("signalr/test", error.SessionId.Value);
+    }
+}

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<SessionOutput>();
+        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<TextOutput>().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<IHost> 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<FakeHubState>();
+
+        var app = builder.Build();
+        app.MapHub<FakeSessionHub>("/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<string> _sessions = [];
+        private readonly ConcurrentDictionary<string, string> _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<ISessionHubClient>
+    {
+        private readonly FakeHubState _state;
+
+        public FakeSessionHub(FakeHubState state)
+        {
+            _state = state;
+        }
+
+        public Task<SessionEnsureResultDto> 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);
+        }
+    }
+}

src/Netclaw.Actors.Tests/Netclaw.Actors.Tests.csproj

@@ -11,6 +11,7 @@
   <ItemGroup>
     <PackageReference Include="Akka.Hosting.TestKit"/>
     <PackageReference Include="Akka.Persistence.Hosting"/>
+    <PackageReference Include="Microsoft.AspNetCore.TestHost" />
     <PackageReference Include="Microsoft.NET.Test.Sdk"/>
     <PackageReference Include="xunit"/>
     <PackageReference Include="xunit.runner.visualstudio"/>

src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs

@@ -434,7 +434,20 @@ public IAsyncEnumerable<ChatResponseUpdate> GetStreamingResponseAsync(
         ChatOptions? options = null,
         CancellationToken cancellationToken = default)
     {
-        throw new NotSupportedException("Streaming not used in tests");
+        return CreateStreamingUpdatesAsync(messages, options, cancellationToken);
+    }
+
+    private async IAsyncEnumerable<ChatResponseUpdate> CreateStreamingUpdatesAsync(
+        IEnumerable<ChatMessage> 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;

src/Netclaw.Actors/Protocol/SessionEnsureResultDto.cs

@@ -0,0 +1,14 @@
+namespace Netclaw.Actors.Protocol;
+
+/// <summary>
+/// Wire-safe response for ensuring a SignalR session binding.
+/// </summary>
+public sealed record SessionEnsureResultDto
+{
+    public required string SessionId { get; init; }
+
+    /// <summary>
+    /// True when a new session was created; false when existing session was reattached.
+    /// </summary>
+    public required bool Created { get; init; }
+}

src/Netclaw.Actors/Protocol/SessionOutput.cs

@@ -28,6 +28,15 @@ public sealed record TextOutput : SessionOutput
     public required string Text { get; init; }
 }
 
+/// <summary>
+/// Incremental text delta from the assistant while a turn is streaming.
+/// Requires <see cref="OutputFilter.Text"/>.
+/// </summary>
+public sealed record TextDeltaOutput : SessionOutput
+{
+    public required string Delta { get; init; }
+}
+
 /// <summary>
 /// Thinking/reasoning tokens from the model (e.g., Claude extended thinking).
 /// Requires <see cref="OutputFilter.Thinking"/>.
@@ -37,6 +46,15 @@ public sealed record ThinkingOutput : SessionOutput
     public required string Text { get; init; }
 }
 
+/// <summary>
+/// Incremental thinking/reasoning delta while a turn is streaming.
+/// Requires <see cref="OutputFilter.Thinking"/>.
+/// </summary>
+public sealed record ThinkingDeltaOutput : SessionOutput
+{
+    public required string Delta { get; init; }
+}
+
 /// <summary>
 /// The model has requested a tool/function call.
 /// Requires <see cref="OutputFilter.ToolCalls"/>.

src/Netclaw.Actors/Protocol/SessionOutputDto.cs

@@ -8,7 +8,7 @@ namespace Netclaw.Actors.Protocol;
 public sealed record SessionOutputDto
 {
     /// <summary>
-    /// 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").
     /// </summary>

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; }
+}
+
+/// <summary>
+/// Incremental streaming delta emitted while an LLM response is in-flight.
+/// </summary>
+internal sealed record LlmResponseDeltaReceived
+{
+    public required AIContent Content { get; init; }
 }
 
 /// <summary>