Phase 1 done

src/Runner.Worker/Dap/DapDebugSession.cs

@@ -0,0 +1,643 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using GitHub.DistributedTask.WebApi;
+using GitHub.Runner.Common;
+using Newtonsoft.Json;
+
+namespace GitHub.Runner.Worker.Dap
+{
+    /// <summary>
+    /// Debug session state machine states.
+    /// </summary>
+    public enum DapSessionState
+    {
+        /// <summary>
+        /// Initial state, waiting for client connection.
+        /// </summary>
+        WaitingForConnection,
+
+        /// <summary>
+        /// Client connected, exchanging capabilities.
+        /// </summary>
+        Initializing,
+
+        /// <summary>
+        /// ConfigurationDone received, ready to debug.
+        /// </summary>
+        Ready,
+
+        /// <summary>
+        /// Paused before or after a step, waiting for user command.
+        /// </summary>
+        Paused,
+
+        /// <summary>
+        /// Executing a step.
+        /// </summary>
+        Running,
+
+        /// <summary>
+        /// Session disconnected or terminated.
+        /// </summary>
+        Terminated
+    }
+
+    /// <summary>
+    /// Commands that can be issued from the debug client.
+    /// </summary>
+    public enum DapCommand
+    {
+        /// <summary>
+        /// Continue execution until end or next breakpoint.
+        /// </summary>
+        Continue,
+
+        /// <summary>
+        /// Execute current step and pause before next.
+        /// </summary>
+        Next,
+
+        /// <summary>
+        /// Pause execution.
+        /// </summary>
+        Pause,
+
+        /// <summary>
+        /// Disconnect from the debug session.
+        /// </summary>
+        Disconnect
+    }
+
+    /// <summary>
+    /// Reasons for stopping/pausing execution.
+    /// </summary>
+    public static class StopReason
+    {
+        public const string Entry = "entry";
+        public const string Step = "step";
+        public const string Breakpoint = "breakpoint";
+        public const string Pause = "pause";
+        public const string Exception = "exception";
+    }
+
+    /// <summary>
+    /// Interface for the DAP debug session.
+    /// Handles debug state, step coordination, and DAP request processing.
+    /// </summary>
+    [ServiceLocator(Default = typeof(DapDebugSession))]
+    public interface IDapDebugSession : IRunnerService
+    {
+        /// <summary>
+        /// Gets whether the debug session is active (initialized and configured).
+        /// </summary>
+        bool IsActive { get; }
+
+        /// <summary>
+        /// Gets the current session state.
+        /// </summary>
+        DapSessionState State { get; }
+
+        /// <summary>
+        /// Sets the DAP server for sending events.
+        /// </summary>
+        /// <param name="server">The DAP server</param>
+        void SetDapServer(IDapServer server);
+
+        /// <summary>
+        /// Handles an incoming DAP request and returns a response.
+        /// </summary>
+        /// <param name="request">The DAP request</param>
+        /// <returns>The DAP response</returns>
+        Task<Response> HandleRequestAsync(Request request);
+
+        /// <summary>
+        /// Called by StepsRunner before a step starts executing.
+        /// May block waiting for debugger commands.
+        /// </summary>
+        /// <param name="step">The step about to execute</param>
+        /// <param name="jobContext">The job execution context</param>
+        /// <param name="isFirstStep">Whether this is the first step in the job</param>
+        /// <returns>Task that completes when execution should continue</returns>
+        Task OnStepStartingAsync(IStep step, IExecutionContext jobContext, bool isFirstStep);
+
+        /// <summary>
+        /// Called by StepsRunner after a step completes.
+        /// </summary>
+        /// <param name="step">The step that completed</param>
+        void OnStepCompleted(IStep step);
+
+        /// <summary>
+        /// Notifies the session that the job has completed.
+        /// </summary>
+        void OnJobCompleted();
+    }
+
+    /// <summary>
+    /// Debug session implementation for handling DAP requests and coordinating
+    /// step execution with the debugger.
+    /// </summary>
+    public sealed class DapDebugSession : RunnerService, IDapDebugSession
+    {
+        // Thread ID for the single job execution thread
+        private const int JobThreadId = 1;
+
+        // Frame ID for the current step
+        private const int CurrentFrameId = 1;
+
+        private IDapServer _server;
+        private DapSessionState _state = DapSessionState.WaitingForConnection;
+        private InitializeRequestArguments _clientCapabilities;
+
+        // Synchronization for step execution
+        private TaskCompletionSource<DapCommand> _commandTcs;
+        private readonly object _stateLock = new object();
+
+        // Current execution context (set during OnStepStartingAsync)
+        private IStep _currentStep;
+        private IExecutionContext _jobContext;
+
+        public bool IsActive => _state == DapSessionState.Ready || _state == DapSessionState.Paused || _state == DapSessionState.Running;
+
+        public DapSessionState State => _state;
+
+        public override void Initialize(IHostContext hostContext)
+        {
+            base.Initialize(hostContext);
+            Trace.Info("DapDebugSession initialized");
+        }
+
+        public void SetDapServer(IDapServer server)
+        {
+            _server = server;
+            Trace.Info("DAP server reference set");
+        }
+
+        public async Task<Response> HandleRequestAsync(Request request)
+        {
+            Trace.Info($"Handling DAP request: {request.Command}");
+
+            try
+            {
+                return request.Command switch
+                {
+                    "initialize" => HandleInitialize(request),
+                    "attach" => HandleAttach(request),
+                    "configurationDone" => HandleConfigurationDone(request),
+                    "disconnect" => HandleDisconnect(request),
+                    "threads" => HandleThreads(request),
+                    "stackTrace" => HandleStackTrace(request),
+                    "scopes" => HandleScopes(request),
+                    "variables" => HandleVariables(request),
+                    "continue" => HandleContinue(request),
+                    "next" => HandleNext(request),
+                    "pause" => HandlePause(request),
+                    "evaluate" => await HandleEvaluateAsync(request),
+                    "setBreakpoints" => HandleSetBreakpoints(request),
+                    "setExceptionBreakpoints" => HandleSetExceptionBreakpoints(request),
+                    _ => CreateErrorResponse($"Unknown command: {request.Command}")
+                };
+            }
+            catch (Exception ex)
+            {
+                Trace.Error($"Error handling request '{request.Command}': {ex}");
+                return CreateErrorResponse(ex.Message);
+            }
+        }
+
+        #region DAP Request Handlers
+
+        private Response HandleInitialize(Request request)
+        {
+            // Parse client capabilities
+            if (request.Arguments != null)
+            {
+                _clientCapabilities = request.Arguments.ToObject<InitializeRequestArguments>();
+                Trace.Info($"Client: {_clientCapabilities.ClientName ?? _clientCapabilities.ClientId ?? "unknown"}");
+            }
+
+            _state = DapSessionState.Initializing;
+
+            // Build our capabilities response
+            var capabilities = new Capabilities
+            {
+                SupportsConfigurationDoneRequest = true,
+                SupportsEvaluateForHovers = true,
+                SupportTerminateDebuggee = true,
+                SupportsTerminateRequest = true,
+                // We don't support these features (yet)
+                SupportsStepBack = false,
+                SupportsSetVariable = false,
+                SupportsRestartFrame = false,
+                SupportsGotoTargetsRequest = false,
+                SupportsStepInTargetsRequest = false,
+                SupportsCompletionsRequest = false,
+                SupportsModulesRequest = false,
+                SupportsFunctionBreakpoints = false,
+                SupportsConditionalBreakpoints = false,
+                SupportsExceptionOptions = false,
+                SupportsValueFormattingOptions = false,
+                SupportsExceptionInfoRequest = false,
+                SupportsDelayedStackTraceLoading = false,
+                SupportsLoadedSourcesRequest = false,
+                SupportsProgressReporting = false,
+                SupportsRunInTerminalRequest = false,
+                SupportsCancelRequest = false,
+            };
+
+            // Queue the initialized event to be sent after the response
+            Task.Run(() =>
+            {
+                // Small delay to ensure response is sent first
+                System.Threading.Thread.Sleep(50);
+                _server?.SendEvent(new Event
+                {
+                    EventType = "initialized"
+                });
+                Trace.Info("Sent initialized event");
+            });
+
+            Trace.Info("Initialize request handled, capabilities sent");
+            return CreateSuccessResponse(capabilities);
+        }
+
+        private Response HandleAttach(Request request)
+        {
+            Trace.Info("Attach request handled");
+            return CreateSuccessResponse(null);
+        }
+
+        private Response HandleConfigurationDone(Request request)
+        {
+            lock (_stateLock)
+            {
+                _state = DapSessionState.Ready;
+            }
+            Trace.Info("Configuration done, debug session is ready");
+
+            // Complete any pending wait for configuration
+            return CreateSuccessResponse(null);
+        }
+
+        private Response HandleDisconnect(Request request)
+        {
+            Trace.Info("Disconnect request received");
+
+            lock (_stateLock)
+            {
+                _state = DapSessionState.Terminated;
+
+                // Release any blocked step execution
+                _commandTcs?.TrySetResult(DapCommand.Disconnect);
+            }
+
+            return CreateSuccessResponse(null);
+        }
+
+        private Response HandleThreads(Request request)
+        {
+            // We have a single thread representing the job execution
+            var body = new ThreadsResponseBody
+            {
+                Threads = new System.Collections.Generic.List<Thread>
+                {
+                    new Thread
+                    {
+                        Id = JobThreadId,
+                        Name = _jobContext != null
+                            ? $"Job: {_jobContext.GetGitHubContext("job") ?? "workflow job"}"
+                            : "Job Thread"
+                    }
+                }
+            };
+
+            return CreateSuccessResponse(body);
+        }
+
+        private Response HandleStackTrace(Request request)
+        {
+            var args = request.Arguments?.ToObject<StackTraceArguments>();
+
+            var frames = new System.Collections.Generic.List<StackFrame>();
+
+            // Add current step as the top frame
+            if (_currentStep != null)
+            {
+                frames.Add(new StackFrame
+                {
+                    Id = CurrentFrameId,
+                    Name = _currentStep.DisplayName ?? "Current Step",
+                    Line = 1,
+                    Column = 1,
+                    PresentationHint = "normal"
+                });
+            }
+            else
+            {
+                frames.Add(new StackFrame
+                {
+                    Id = CurrentFrameId,
+                    Name = "(no step executing)",
+                    Line = 1,
+                    Column = 1,
+                    PresentationHint = "subtle"
+                });
+            }
+
+            // TODO: In Phase 2, add completed steps as additional frames
+
+            var body = new StackTraceResponseBody
+            {
+                StackFrames = frames,
+                TotalFrames = frames.Count
+            };
+
+            return CreateSuccessResponse(body);
+        }
+
+        private Response HandleScopes(Request request)
+        {
+            // Stub implementation - Phase 2 will populate with actual contexts
+            var body = new ScopesResponseBody
+            {
+                Scopes = new System.Collections.Generic.List<Scope>
+                {
+                    new Scope { Name = "github", VariablesReference = 1, Expensive = false },
+                    new Scope { Name = "env", VariablesReference = 2, Expensive = false },
+                    new Scope { Name = "runner", VariablesReference = 3, Expensive = false },
+                    new Scope { Name = "job", VariablesReference = 4, Expensive = false },
+                    new Scope { Name = "steps", VariablesReference = 5, Expensive = false },
+                    new Scope { Name = "secrets", VariablesReference = 6, Expensive = false, PresentationHint = "registers" },
+                }
+            };
+
+            return CreateSuccessResponse(body);
+        }
+
+        private Response HandleVariables(Request request)
+        {
+            // Stub implementation - Phase 2 will populate with actual variable values
+            var args = request.Arguments?.ToObject<VariablesArguments>();
+            var variablesRef = args?.VariablesReference ?? 0;
+
+            var body = new VariablesResponseBody
+            {
+                Variables = new System.Collections.Generic.List<Variable>
+                {
+                    new Variable
+                    {
+                        Name = "(stub)",
+                        Value = $"Variables for scope {variablesRef} will be implemented in Phase 2",
+                        Type = "string",
+                        VariablesReference = 0
+                    }
+                }
+            };
+
+            return CreateSuccessResponse(body);
+        }
+
+        private Response HandleContinue(Request request)
+        {
+            Trace.Info("Continue command received");
+
+            lock (_stateLock)
+            {
+                if (_state == DapSessionState.Paused)
+                {
+                    _state = DapSessionState.Running;
+                    _commandTcs?.TrySetResult(DapCommand.Continue);
+                }
+            }
+
+            return CreateSuccessResponse(new ContinueResponseBody
+            {
+                AllThreadsContinued = true
+            });
+        }
+
+        private Response HandleNext(Request request)
+        {
+            Trace.Info("Next (step over) command received");
+
+            lock (_stateLock)
+            {
+                if (_state == DapSessionState.Paused)
+                {
+                    _state = DapSessionState.Running;
+                    _commandTcs?.TrySetResult(DapCommand.Next);
+                }
+            }
+
+            return CreateSuccessResponse(null);
+        }
+
+        private Response HandlePause(Request request)
+        {
+            Trace.Info("Pause command received");
+
+            // The runner will pause at the next step boundary
+            lock (_stateLock)
+            {
+                // Just acknowledge - actual pause happens at step boundary
+            }
+
+            return CreateSuccessResponse(null);
+        }
+
+        private async Task<Response> HandleEvaluateAsync(Request request)
+        {
+            var args = request.Arguments?.ToObject<EvaluateArguments>();
+            var expression = args?.Expression ?? "";
+            var context = args?.Context ?? "hover";
+
+            Trace.Info($"Evaluate: '{expression}' (context: {context})");
+
+            // Stub implementation - Phase 4 will implement expression evaluation
+            var body = new EvaluateResponseBody
+            {
+                Result = $"(evaluation of '{expression}' will be implemented in Phase 4)",
+                Type = "string",
+                VariablesReference = 0
+            };
+
+            return CreateSuccessResponse(body);
+        }
+
+        private Response HandleSetBreakpoints(Request request)
+        {
+            // Stub - breakpoints not implemented in demo
+            Trace.Info("SetBreakpoints request (not implemented)");
+
+            return CreateSuccessResponse(new
+            {
+                breakpoints = new object[0]
+            });
+        }
+
+        private Response HandleSetExceptionBreakpoints(Request request)
+        {
+            // Stub - exception breakpoints not implemented in demo
+            Trace.Info("SetExceptionBreakpoints request (not implemented)");
+
+            return CreateSuccessResponse(new
+            {
+                breakpoints = new object[0]
+            });
+        }
+
+        #endregion
+
+        #region Step Coordination (called by StepsRunner)
+
+        public async Task OnStepStartingAsync(IStep step, IExecutionContext jobContext, bool isFirstStep)
+        {
+            if (!IsActive)
+            {
+                return;
+            }
+
+            _currentStep = step;
+            _jobContext = jobContext;
+
+            var reason = isFirstStep ? StopReason.Entry : StopReason.Step;
+            var description = isFirstStep
+                ? $"Stopped at job entry: {step.DisplayName}"
+                : $"Stopped before step: {step.DisplayName}";
+
+            Trace.Info($"Step starting: {step.DisplayName} (reason: {reason})");
+
+            // Send stopped event to debugger
+            _server?.SendEvent(new Event
+            {
+                EventType = "stopped",
+                Body = new StoppedEventBody
+                {
+                    Reason = reason,
+                    Description = description,
+                    ThreadId = JobThreadId,
+                    AllThreadsStopped = true
+                }
+            });
+
+            // Wait for debugger command
+            await WaitForCommandAsync();
+        }
+
+        public void OnStepCompleted(IStep step)
+        {
+            if (!IsActive)
+            {
+                return;
+            }
+
+            Trace.Info($"Step completed: {step.DisplayName}, result: {step.ExecutionContext?.Result}");
+
+            // The step context will be available for inspection
+            // Future: could pause here if "pause after step" is enabled
+        }
+
+        public void OnJobCompleted()
+        {
+            if (!IsActive)
+            {
+                return;
+            }
+
+            Trace.Info("Job completed, sending terminated event");
+
+            lock (_stateLock)
+            {
+                _state = DapSessionState.Terminated;
+            }
+
+            // Send terminated event
+            _server?.SendEvent(new Event
+            {
+                EventType = "terminated",
+                Body = new TerminatedEventBody()
+            });
+
+            // Send exited event
+            var exitCode = _jobContext?.Result == GitHub.DistributedTask.WebApi.TaskResult.Succeeded ? 0 : 1;
+            _server?.SendEvent(new Event
+            {
+                EventType = "exited",
+                Body = new ExitedEventBody
+                {
+                    ExitCode = exitCode
+                }
+            });
+        }
+
+        private async Task WaitForCommandAsync()
+        {
+            lock (_stateLock)
+            {
+                _state = DapSessionState.Paused;
+                _commandTcs = new TaskCompletionSource<DapCommand>(TaskCreationOptions.RunContinuationsAsynchronously);
+            }
+
+            Trace.Info("Waiting for debugger command...");
+
+            var command = await _commandTcs.Task;
+
+            Trace.Info($"Received command: {command}");
+
+            lock (_stateLock)
+            {
+                if (_state == DapSessionState.Paused)
+                {
+                    _state = DapSessionState.Running;
+                }
+            }
+
+            // Send continued event
+            if (command == DapCommand.Continue || command == DapCommand.Next)
+            {
+                _server?.SendEvent(new Event
+                {
+                    EventType = "continued",
+                    Body = new ContinuedEventBody
+                    {
+                        ThreadId = JobThreadId,
+                        AllThreadsContinued = true
+                    }
+                });
+            }
+        }
+
+        #endregion
+
+        #region Response Helpers
+
+        private Response CreateSuccessResponse(object body)
+        {
+            return new Response
+            {
+                Success = true,
+                Body = body
+            };
+        }
+
+        private Response CreateErrorResponse(string message)
+        {
+            return new Response
+            {
+                Success = false,
+                Message = message,
+                Body = new ErrorResponseBody
+                {
+                    Error = new Message
+                    {
+                        Id = 1,
+                        Format = message,
+                        ShowUser = true
+                    }
+                }
+            };
+        }
+
+        #endregion
+    }
+}

src/Runner.Worker/Dap/DapServer.cs

@@ -0,0 +1,480 @@
+using System;
+using System.IO;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using GitHub.Runner.Common;
+using Newtonsoft.Json;
+
+namespace GitHub.Runner.Worker.Dap
+{
+    /// <summary>
+    /// DAP Server interface for handling Debug Adapter Protocol connections.
+    /// </summary>
+    [ServiceLocator(Default = typeof(DapServer))]
+    public interface IDapServer : IRunnerService, IDisposable
+    {
+        /// <summary>
+        /// Starts the DAP TCP server on the specified port.
+        /// </summary>
+        /// <param name="port">The port to listen on (default: 4711)</param>
+        /// <param name="cancellationToken">Cancellation token</param>
+        Task StartAsync(int port, CancellationToken cancellationToken);
+
+        /// <summary>
+        /// Blocks until a debug client connects.
+        /// </summary>
+        /// <param name="cancellationToken">Cancellation token</param>
+        Task WaitForConnectionAsync(CancellationToken cancellationToken);
+
+        /// <summary>
+        /// Stops the DAP server and closes all connections.
+        /// </summary>
+        Task StopAsync();
+
+        /// <summary>
+        /// Sets the debug session that will handle DAP requests.
+        /// </summary>
+        /// <param name="session">The debug session</param>
+        void SetSession(IDapDebugSession session);
+
+        /// <summary>
+        /// Sends an event to the connected debug client.
+        /// </summary>
+        /// <param name="evt">The event to send</param>
+        void SendEvent(Event evt);
+
+        /// <summary>
+        /// Gets whether a debug client is currently connected.
+        /// </summary>
+        bool IsConnected { get; }
+    }
+
+    /// <summary>
+    /// TCP server implementation of the Debug Adapter Protocol.
+    /// Handles message framing (Content-Length headers) and JSON serialization.
+    /// </summary>
+    public sealed class DapServer : RunnerService, IDapServer
+    {
+        private const string ContentLengthHeader = "Content-Length: ";
+        private const string HeaderTerminator = "\r\n\r\n";
+
+        private TcpListener _listener;
+        private TcpClient _client;
+        private NetworkStream _stream;
+        private IDapDebugSession _session;
+        private CancellationTokenSource _cts;
+        private Task _messageLoopTask;
+        private TaskCompletionSource<bool> _connectionTcs;
+        private int _nextSeq = 1;
+        private readonly object _sendLock = new object();
+        private bool _disposed = false;
+
+        public bool IsConnected => _client?.Connected == true;
+
+        public override void Initialize(IHostContext hostContext)
+        {
+            base.Initialize(hostContext);
+            Trace.Info("DapServer initialized");
+        }
+
+        public void SetSession(IDapDebugSession session)
+        {
+            _session = session;
+            Trace.Info("Debug session set");
+        }
+
+        public async Task StartAsync(int port, CancellationToken cancellationToken)
+        {
+            Trace.Info($"Starting DAP server on port {port}");
+
+            _cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+            _connectionTcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+            try
+            {
+                _listener = new TcpListener(IPAddress.Loopback, port);
+                _listener.Start();
+                Trace.Info($"DAP server listening on 127.0.0.1:{port}");
+
+                // Start accepting connections in the background
+                _ = AcceptConnectionAsync(_cts.Token);
+            }
+            catch (Exception ex)
+            {
+                Trace.Error($"Failed to start DAP server: {ex.Message}");
+                throw;
+            }
+
+            await Task.CompletedTask;
+        }
+
+        private async Task AcceptConnectionAsync(CancellationToken cancellationToken)
+        {
+            try
+            {
+                Trace.Info("Waiting for debug client connection...");
+
+                // Use cancellation-aware accept
+                using (cancellationToken.Register(() => _listener?.Stop()))
+                {
+                    _client = await _listener.AcceptTcpClientAsync();
+                }
+
+                if (cancellationToken.IsCancellationRequested)
+                {
+                    return;
+                }
+
+                _stream = _client.GetStream();
+                var remoteEndPoint = _client.Client.RemoteEndPoint;
+                Trace.Info($"Debug client connected from {remoteEndPoint}");
+
+                // Signal that connection is established
+                _connectionTcs.TrySetResult(true);
+
+                // Start processing messages
+                _messageLoopTask = ProcessMessagesAsync(_cts.Token);
+            }
+            catch (ObjectDisposedException) when (cancellationToken.IsCancellationRequested)
+            {
+                // Expected when cancellation stops the listener
+                Trace.Info("Connection accept cancelled");
+                _connectionTcs.TrySetCanceled();
+            }
+            catch (SocketException ex) when (cancellationToken.IsCancellationRequested)
+            {
+                // Expected when cancellation stops the listener
+                Trace.Info($"Connection accept cancelled: {ex.Message}");
+                _connectionTcs.TrySetCanceled();
+            }
+            catch (Exception ex)
+            {
+                Trace.Error($"Error accepting connection: {ex.Message}");
+                _connectionTcs.TrySetException(ex);
+            }
+        }
+
+        public async Task WaitForConnectionAsync(CancellationToken cancellationToken)
+        {
+            Trace.Info("Waiting for debug client to connect...");
+
+            using (cancellationToken.Register(() => _connectionTcs.TrySetCanceled()))
+            {
+                await _connectionTcs.Task;
+            }
+
+            Trace.Info("Debug client connected");
+        }
+
+        public async Task StopAsync()
+        {
+            Trace.Info("Stopping DAP server");
+
+            _cts?.Cancel();
+
+            // Wait for message loop to complete
+            if (_messageLoopTask != null)
+            {
+                try
+                {
+                    await _messageLoopTask;
+                }
+                catch (OperationCanceledException)
+                {
+                    // Expected
+                }
+                catch (Exception ex)
+                {
+                    Trace.Warning($"Message loop ended with error: {ex.Message}");
+                }
+            }
+
+            // Clean up resources
+            _stream?.Close();
+            _client?.Close();
+            _listener?.Stop();
+
+            Trace.Info("DAP server stopped");
+        }
+
+        public void SendEvent(Event evt)
+        {
+            if (!IsConnected)
+            {
+                Trace.Warning($"Cannot send event '{evt.EventType}': no client connected");
+                return;
+            }
+
+            try
+            {
+                lock (_sendLock)
+                {
+                    evt.Seq = _nextSeq++;
+                    SendMessageInternal(evt);
+                }
+                Trace.Info($"Sent event: {evt.EventType}");
+            }
+            catch (Exception ex)
+            {
+                Trace.Error($"Failed to send event '{evt.EventType}': {ex.Message}");
+            }
+        }
+
+        private async Task ProcessMessagesAsync(CancellationToken cancellationToken)
+        {
+            Trace.Info("Starting DAP message processing loop");
+
+            try
+            {
+                while (!cancellationToken.IsCancellationRequested && IsConnected)
+                {
+                    var json = await ReadMessageAsync(cancellationToken);
+                    if (json == null)
+                    {
+                        Trace.Info("Client disconnected (end of stream)");
+                        break;
+                    }
+
+                    await ProcessMessageAsync(json, cancellationToken);
+                }
+            }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                Trace.Info("Message processing cancelled");
+            }
+            catch (IOException ex)
+            {
+                Trace.Info($"Connection closed: {ex.Message}");
+            }
+            catch (Exception ex)
+            {
+                Trace.Error($"Error in message loop: {ex}");
+            }
+
+            Trace.Info("DAP message processing loop ended");
+        }
+
+        private async Task ProcessMessageAsync(string json, CancellationToken cancellationToken)
+        {
+            Request request = null;
+            try
+            {
+                // Parse the incoming message
+                request = JsonConvert.DeserializeObject<Request>(json);
+                if (request == null || request.Type != "request")
+                {
+                    Trace.Warning($"Received non-request message: {json}");
+                    return;
+                }
+
+                Trace.Info($"Received request: seq={request.Seq}, command={request.Command}");
+
+                // Dispatch to session for handling
+                if (_session == null)
+                {
+                    Trace.Error("No debug session configured");
+                    SendErrorResponse(request, "No debug session configured");
+                    return;
+                }
+
+                var response = await _session.HandleRequestAsync(request);
+                response.RequestSeq = request.Seq;
+                response.Command = request.Command;
+                response.Type = "response";
+
+                lock (_sendLock)
+                {
+                    response.Seq = _nextSeq++;
+                    SendMessageInternal(response);
+                }
+
+                Trace.Info($"Sent response: seq={response.Seq}, command={response.Command}, success={response.Success}");
+            }
+            catch (JsonException ex)
+            {
+                Trace.Error($"Failed to parse request: {ex.Message}");
+                Trace.Error($"JSON: {json}");
+            }
+            catch (Exception ex)
+            {
+                Trace.Error($"Error processing request: {ex}");
+                if (request != null)
+                {
+                    SendErrorResponse(request, ex.Message);
+                }
+            }
+        }
+
+        private void SendErrorResponse(Request request, string message)
+        {
+            var response = new Response
+            {
+                Type = "response",
+                RequestSeq = request.Seq,
+                Command = request.Command,
+                Success = false,
+                Message = message,
+                Body = new ErrorResponseBody
+                {
+                    Error = new Message
+                    {
+                        Id = 1,
+                        Format = message,
+                        ShowUser = true
+                    }
+                }
+            };
+
+            lock (_sendLock)
+            {
+                response.Seq = _nextSeq++;
+                SendMessageInternal(response);
+            }
+        }
+
+        /// <summary>
+        /// Reads a DAP message from the stream.
+        /// DAP uses HTTP-like message framing: Content-Length: N\r\n\r\n{json}
+        /// </summary>
+        private async Task<string> ReadMessageAsync(CancellationToken cancellationToken)
+        {
+            // Read headers until we find Content-Length
+            var headerBuilder = new StringBuilder();
+            int contentLength = -1;
+
+            while (true)
+            {
+                var line = await ReadLineAsync(cancellationToken);
+                if (line == null)
+                {
+                    // End of stream
+                    return null;
+                }
+
+                if (line.Length == 0)
+                {
+                    // Empty line marks end of headers
+                    break;
+                }
+
+                headerBuilder.AppendLine(line);
+
+                if (line.StartsWith(ContentLengthHeader, StringComparison.OrdinalIgnoreCase))
+                {
+                    var lengthStr = line.Substring(ContentLengthHeader.Length).Trim();
+                    if (!int.TryParse(lengthStr, out contentLength))
+                    {
+                        throw new InvalidDataException($"Invalid Content-Length: {lengthStr}");
+                    }
+                }
+            }
+
+            if (contentLength < 0)
+            {
+                throw new InvalidDataException("Missing Content-Length header");
+            }
+
+            // Read the JSON body
+            var buffer = new byte[contentLength];
+            var totalRead = 0;
+            while (totalRead < contentLength)
+            {
+                var bytesRead = await _stream.ReadAsync(buffer, totalRead, contentLength - totalRead, cancellationToken);
+                if (bytesRead == 0)
+                {
+                    throw new EndOfStreamException("Connection closed while reading message body");
+                }
+                totalRead += bytesRead;
+            }
+
+            var json = Encoding.UTF8.GetString(buffer);
+            Trace.Verbose($"Received: {json}");
+            return json;
+        }
+
+        /// <summary>
+        /// Reads a line from the stream (terminated by \r\n).
+        /// </summary>
+        private async Task<string> ReadLineAsync(CancellationToken cancellationToken)
+        {
+            var lineBuilder = new StringBuilder();
+            var buffer = new byte[1];
+            var previousWasCr = false;
+
+            while (true)
+            {
+                var bytesRead = await _stream.ReadAsync(buffer, 0, 1, cancellationToken);
+                if (bytesRead == 0)
+                {
+                    // End of stream
+                    return lineBuilder.Length > 0 ? lineBuilder.ToString() : null;
+                }
+
+                var c = (char)buffer[0];
+
+                if (c == '\n' && previousWasCr)
+                {
+                    // Found \r\n, return the line (without the \r)
+                    if (lineBuilder.Length > 0 && lineBuilder[lineBuilder.Length - 1] == '\r')
+                    {
+                        lineBuilder.Length--;
+                    }
+                    return lineBuilder.ToString();
+                }
+
+                previousWasCr = (c == '\r');
+                lineBuilder.Append(c);
+            }
+        }
+
+        /// <summary>
+        /// Sends a DAP message to the stream with Content-Length framing.
+        /// Must be called within the _sendLock.
+        /// </summary>
+        private void SendMessageInternal(ProtocolMessage message)
+        {
+            var json = JsonConvert.SerializeObject(message, new JsonSerializerSettings
+            {
+                NullValueHandling = NullValueHandling.Ignore
+            });
+
+            var bodyBytes = Encoding.UTF8.GetBytes(json);
+            var header = $"Content-Length: {bodyBytes.Length}\r\n\r\n";
+            var headerBytes = Encoding.UTF8.GetBytes(header);
+
+            _stream.Write(headerBytes, 0, headerBytes.Length);
+            _stream.Write(bodyBytes, 0, bodyBytes.Length);
+            _stream.Flush();
+
+            Trace.Verbose($"Sent: {json}");
+        }
+
+        public void Dispose()
+        {
+            Dispose(true);
+            GC.SuppressFinalize(this);
+        }
+
+        private void Dispose(bool disposing)
+        {
+            if (_disposed)
+            {
+                return;
+            }
+
+            if (disposing)
+            {
+                _cts?.Cancel();
+                _stream?.Dispose();
+                _client?.Dispose();
+                _listener?.Stop();
+                _cts?.Dispose();
+            }
+
+            _disposed = true;
+        }
+    }
+}