Refactor LitAgentRunnerTests to support configurable chat client and options

- Updated CreateAgent method to accept optional chatClient, options, and loggerFactory parameters.
- Modified test methods to utilize enqueued rollouts and validate spans.
- Added a new test for retrying failed rollouts with a flaky chat client.

Enhance InMemoryLightningStoreTests with additional functionality

- Added tests for StartRollout, StartAttempt, WaitForRollouts, and GetNextSpanSequenceId methods.
- Improved QueryRollouts to filter by status.

Implement TraceToTripletAdapter for converting spans to triplet trajectories

- Created TracerTraceToTripletAdapter class to handle span conversion.
- Added logic for reward matching policies and deduplication of LLM calls.

Add unit tests for TracerTraceToTripletAdapter

- Implemented tests to validate token and reward projection, deduplication, and reward policies.

Introduce FlakyChatClient for simulating intermittent failures in tests

- Created FlakyChatClient to simulate failures before returning a successful response.

Author: KSemenenko

MIGRATION_PLAN.md

@@ -26,18 +26,18 @@ This plan tracks parity work between `external/microsoft-agent-lightning` (Pytho
 
 | Adapter | Python Source | Status | Notes |
 | --- | --- | --- | --- |
-| Trace → messages | `adapter/messages.py` | 🚧 | Convert spans to chat history for `IChatClient` |
-| Trace → triplets | `adapter/triplet.py` | 🚧 | Build reward-aware triplets from spans |
+| Trace → messages | `adapter/messages.py` | ✅ | `TraceToMessagesAdapter` translates GenAI spans into OpenAI chat payloads |
+| Trace → triplets | `adapter/triplet.py` | ✅ | `TracerTraceToTripletAdapter` exports triplets with reward policies |
 | OTEL trace adapter | `adapter/base.py` | 🚧 | Hook Activity -> SpanModel bridging |
 
 ## Execution & Store Layers
 
 | Component | Python Source | Status | Notes |
 | --- | --- | --- | --- |
-| LightningStore (async) | `store/base.py` | 🚧 | Async contract mirroring enqueue/start/query APIs |
-| In-memory store | `store/memory.py` | 🚧 | Production-grade concurrency port |
+| LightningStore (async) | `store/base.py` | ✅ | `ILightningStore` exposes start/enqueue/start-attempt, span sequencing, and wait semantics |
+| In-memory store | `store/memory.py` | ✅ | Expanded store handles attempts, spans, resources, and polling waits with thread-safe state |
 | Client/server bridge | `store/client_server.py` | ❓ | Decide ASP.NET hosting approach |
-| Runner execution strategies | `execution/*` | 🚧 | Channel/task-based equivalents |
+| Runner execution strategies | `execution/*` | 🚧 | C# runner tracks worker ids, retries via store requeue; parallel orchestration still pending |
 
 ## Algorithms & Training Pipelines
 
@@ -97,11 +97,10 @@ This plan tracks parity work between `external/microsoft-agent-lightning` (Pytho
 
 ## Near-Term Priorities
 
-1. Port trace adapters (messages, triplets) to convert spans into actionable payloads.
-2. Implement store abstractions (interface + in-memory implementation) to unblock runner tests.
-3. Port runner orchestration and cover end-to-end execution (agent + store + adapters).
-4. Reproduce key Python fixtures/tests for adapters and store logic.
-5. Plan hosting story for LLM proxy / legacy endpoints (document decisions).
+1. Expand runner execution strategies (parallel workers, retries, resource coordination).
+2. Reproduce key Python fixtures/tests for adapters, store logic, and integration flows.
+3. Plan hosting story for LLM proxy / legacy endpoints and capture decisions.
+4. Map persistence backends beyond in-memory (client/server bridge, durable stores).
 
 ## Tracking Guidance
 

src/ManagedCode.AgentLightning.AgentRuntime/Runner/LitAgentRunner.cs

@@ -1,3 +1,6 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
 using ManagedCode.AgentLightning.Core.Adapters;
 using ManagedCode.AgentLightning.Core.Models;
 using ManagedCode.AgentLightning.Core.Stores;
@@ -15,6 +18,7 @@ public sealed class LitAgentRunner
     private readonly ILightningStore _store;
     private readonly ILogger<LitAgentRunner> _logger;
     private readonly TimeProvider _timeProvider;
+    private string? _workerId;
 
     public LitAgentRunner(
         LightningAgent agent,
@@ -37,7 +41,7 @@ public async Task<IReadOnlyList<LightningExecutionResult>> RunBatchAsync(int exp
 
         var results = new List<LightningExecutionResult>(expectedRollouts);
 
-        for (var i = 0; i < expectedRollouts; i++)
+        while (results.Count < expectedRollouts && !cancellationToken.IsCancellationRequested)
         {
             var attempted = await _store.DequeueRolloutAsync(cancellationToken).ConfigureAwait(false);
             if (attempted is null)
@@ -47,29 +51,87 @@ public async Task<IReadOnlyList<LightningExecutionResult>> RunBatchAsync(int exp
 
             try
             {
+                var workerId = GetWorkerId();
+                attempted.Attempt.AttachWorker(workerId);
+                attempted.Attempt.UpdateStatus(AttemptStatus.Running);
+                attempted.Attempt.Touch(_timeProvider.GetUtcNow());
+                await _store.UpdateAttemptAsync(attempted.Attempt, cancellationToken).ConfigureAwait(false);
+
                 var execution = await _agent.ExecuteAsync(attempted.Rollout.Input, cancellationToken).ConfigureAwait(false);
                 results.Add(execution);
 
                 attempted.Attempt.UpdateStatus(execution.Attempt.Status, execution.Attempt.EndTime ?? _timeProvider.GetUtcNow());
                 await _store.UpdateAttemptAsync(attempted.Attempt, cancellationToken).ConfigureAwait(false);
-                await _store.UpdateRolloutStatusAsync(attempted.Rollout.RolloutId, execution.Rollout.Status, execution.Rollout.EndTime, cancellationToken).ConfigureAwait(false);
 
-                var span = BuildSpanFromResult(attempted, execution);
-                await _store.AddSpanAsync(attempted.Rollout.RolloutId, attempted.Attempt.AttemptId, span, cancellationToken).ConfigureAwait(false);
+                if (execution.Attempt.Status == AttemptStatus.Succeeded)
+                {
+                    await _store.UpdateRolloutStatusAsync(
+                        attempted.Rollout.RolloutId,
+                        execution.Rollout.Status,
+                        execution.Rollout.EndTime ?? _timeProvider.GetUtcNow(),
+                        cancellationToken).ConfigureAwait(false);
+
+                    var sequenceId = await _store.GetNextSpanSequenceIdAsync(
+                        attempted.Rollout.RolloutId,
+                        attempted.Attempt.AttemptId,
+                        cancellationToken).ConfigureAwait(false);
+
+                    var span = BuildSpanFromResult(attempted, execution, sequenceId);
+                    await _store.AddSpanAsync(span, cancellationToken).ConfigureAwait(false);
+                }
+                else
+                {
+                    await HandleAttemptFailureAsync(attempted, cancellationToken).ConfigureAwait(false);
+                }
             }
             catch (Exception ex) when (!cancellationToken.IsCancellationRequested)
             {
                 attempted.Attempt.UpdateStatus(AttemptStatus.Failed, _timeProvider.GetUtcNow());
                 await _store.UpdateAttemptAsync(attempted.Attempt, cancellationToken).ConfigureAwait(false);
-                await _store.UpdateRolloutStatusAsync(attempted.Rollout.RolloutId, RolloutStatus.Failed, _timeProvider.GetUtcNow(), cancellationToken).ConfigureAwait(false);
+                await HandleAttemptFailureAsync(attempted, cancellationToken).ConfigureAwait(false);
                 _logger.LogError(ex, "Runner failed while executing rollout {RolloutId}.", attempted.Rollout.RolloutId);
             }
         }
 
         return results;
     }
 
-    private static SpanModel BuildSpanFromResult(AttemptedRollout attempted, LightningExecutionResult execution)
+    private string GetWorkerId() =>
+        _workerId ??= $"runner-{Environment.CurrentManagedThreadId}";
+
+    private async Task HandleAttemptFailureAsync(AttemptedRollout attempted, CancellationToken cancellationToken)
+    {
+        var shouldRetry = ShouldRetry(attempted.Rollout, attempted.Attempt);
+        var status = shouldRetry ? RolloutStatus.Requeuing : RolloutStatus.Failed;
+        var endTime = shouldRetry ? (DateTimeOffset?)null : _timeProvider.GetUtcNow();
+
+        await _store.UpdateRolloutStatusAsync(
+            attempted.Rollout.RolloutId,
+            status,
+            endTime,
+            cancellationToken).ConfigureAwait(false);
+
+        if (shouldRetry)
+        {
+            _logger.LogInformation("Rollout {RolloutId} requeued after attempt {AttemptId} failed.", attempted.Rollout.RolloutId, attempted.Attempt.AttemptId);
+        }
+        else
+        {
+            _logger.LogWarning("Rollout {RolloutId} failed after attempt {AttemptId}.", attempted.Rollout.RolloutId, attempted.Attempt.AttemptId);
+        }
+    }
+
+    private static bool ShouldRetry(Rollout rollout, Attempt attempt)
+    {
+        if (attempt.SequenceId >= rollout.Config.MaxAttempts)
+        {
+            return false;
+        }
+
+        return rollout.Config.RetryOn.Contains(attempt.Status);
+    }
+
+    private static SpanModel BuildSpanFromResult(AttemptedRollout attempted, LightningExecutionResult execution, int sequenceId)
     {
         var attributes = new Dictionary<string, object?>(StringComparer.Ordinal);
         if (execution.Triplet.Prompt is IEnumerable<object?> prompts)
@@ -103,7 +165,7 @@ private static SpanModel BuildSpanFromResult(AttemptedRollout attempted, Lightni
             attributes,
             rolloutId: attempted.Rollout.RolloutId,
             attemptId: attempted.Attempt.AttemptId,
-            sequenceId: attempted.Attempt.SequenceId,
+            sequenceId: sequenceId,
             name: "agentlightning.completion",
             startTime: attempted.Attempt.StartTime.ToUnixTimeSeconds(),
             endTime: attempted.Attempt.EndTime?.ToUnixTimeSeconds());

src/ManagedCode.AgentLightning.AgentRuntime/Trainer/Trainer.cs

@@ -40,16 +40,12 @@ public async Task<IReadOnlyList<LightningExecutionResult>> TrainAsync(IEnumerabl
             return Array.Empty<LightningExecutionResult>();
         }
 
-        var now = _timeProvider.GetUtcNow();
         foreach (var (payload, index) in taskList.Select((task, index) => (task, index)))
         {
-            var rollout = new Rollout(
-                rolloutId: $"rollout-{now.ToUnixTimeMilliseconds()}-{index}",
-                input: payload,
-                startTime: now,
-                config: new RolloutConfig().Normalize());
-
-            await _store.EnqueueRolloutAsync(rollout, cancellationToken).ConfigureAwait(false);
+            await _store.EnqueueRolloutAsync(
+                payload,
+                config: new RolloutConfig(),
+                cancellationToken: cancellationToken).ConfigureAwait(false);
         }
 
         _logger.LogInformation("Enqueued {Count} rollouts for training.", taskList.Count);