BaseOrchestrator fixed

Author: Kanchan-Microsoft

App/kernel-memory/service/Core/Handlers/DeleteDocumentHandler.cs

@@ -37,7 +37,9 @@ public DeleteDocumentHandler(
     public async Task<(bool success, DataPipeline updatedPipeline)> InvokeAsync(
         DataPipeline pipeline, CancellationToken cancellationToken = default)
     {
-        this._log.LogDebug("Deleting document, pipeline '{0}/{1}'", pipeline.Index, pipeline.DocumentId);
+        this._log.LogDebug("Deleting document, pipeline '{0}/{1}'", 
+            pipeline.Index.Replace("\r", string.Empty).Replace("\n", string.Empty),
+            pipeline.DocumentId.Replace("\r", string.Empty).Replace("\n", string.Empty));
 
         // Delete embeddings
         foreach (IMemoryDb db in this._memoryDbs)

App/kernel-memory/service/Core/Handlers/DeleteGeneratedFilesHandler.cs

@@ -32,7 +32,9 @@ public DeleteGeneratedFilesHandler(
     public async Task<(bool success, DataPipeline updatedPipeline)> InvokeAsync(
         DataPipeline pipeline, CancellationToken cancellationToken = default)
     {
-        this._log.LogDebug("Deleting generated files, pipeline '{0}/{1}'", pipeline.Index, pipeline.DocumentId);
+        this._log.LogDebug("Deleting generated files, pipeline '{0}/{1}'", 
+            pipeline.Index.Replace("\r", string.Empty).Replace("\n", string.Empty),
+            pipeline.DocumentId.Replace("\r", string.Empty).Replace("\n", string.Empty));
 
         // Delete files, leaving the status file
         await this._documentStorage.EmptyDocumentDirectoryAsync(

App/kernel-memory/service/Core/Handlers/DeleteIndexHandler.cs

@@ -37,7 +37,9 @@ public DeleteIndexHandler(
     public async Task<(bool success, DataPipeline updatedPipeline)> InvokeAsync(
         DataPipeline pipeline, CancellationToken cancellationToken = default)
     {
-        this._log.LogDebug("Deleting index, pipeline '{0}/{1}'", pipeline.Index, pipeline.DocumentId);
+        this._log.LogDebug("Deleting index, pipeline '{0}/{1}'", 
+            pipeline.Index.Replace("\r", string.Empty).Replace("\n", string.Empty),
+            pipeline.DocumentId.Replace("\r", string.Empty).Replace("\n", string.Empty));
 
         // Delete index from vector storage
         foreach (IMemoryDb db in this._memoryDbs)

App/kernel-memory/service/Core/Handlers/SummarizationParallelHandler.cs

@@ -56,7 +56,9 @@ public SummarizationParallelHandler(
     public async Task<(bool success, DataPipeline updatedPipeline)> InvokeAsync(
         DataPipeline pipeline, CancellationToken cancellationToken = default)
     {
-        this._log.LogDebug("Generating summary, pipeline '{0}/{1}'", pipeline.Index, pipeline.DocumentId);
+        this._log.LogDebug("Generating summary, pipeline '{0}/{1}'", 
+            pipeline.Index.Replace("\r", string.Empty).Replace("\n", string.Empty),
+            pipeline.DocumentId.Replace("\r", string.Empty).Replace("\n", string.Empty));
 
         foreach (DataPipeline.FileDetails uploadedFile in pipeline.Files)
         {

App/kernel-memory/service/Core/Pipeline/BaseOrchestrator.cs

@@ -428,7 +428,10 @@ protected async Task UploadFilesAsync(DataPipeline currentPipeline, Cancellation
     /// <param name="cancellationToken">Task cancellation token</param>
     protected async Task UpdatePipelineStatusAsync(DataPipeline pipeline, CancellationToken cancellationToken)
     {
-        this.Log.LogDebug("Saving pipeline status to '{0}/{1}/{2}'", pipeline.Index, pipeline.DocumentId, Constants.PipelineStatusFilename);
+        this.Log.LogDebug("Saving pipeline status to '{0}/{1}/{2}'", 
+            pipeline.Index.Replace("\r", string.Empty).Replace("\n", string.Empty),
+            pipeline.DocumentId.Replace("\r", string.Empty).Replace("\n", string.Empty), 
+            Constants.PipelineStatusFilename);
         try
         {
             await this._documentStorage.WriteFileAsync(
@@ -453,7 +456,10 @@ protected static string ToJson(object data, bool indented = false)
 
     private async Task UploadFormFilesAsync(DataPipeline pipeline, CancellationToken cancellationToken)
     {
-        this.Log.LogDebug("Uploading {0} files, pipeline '{1}/{2}'", pipeline.FilesToUpload.Count, pipeline.Index, pipeline.DocumentId);
+        this.Log.LogDebug("Uploading {0} files, pipeline '{1}/{2}'", 
+            pipeline.FilesToUpload.Count, 
+            pipeline.Index.Replace("\r", string.Empty).Replace("\n", string.Empty),
+            pipeline.DocumentId.Replace("\r", string.Empty).Replace("\n", string.Empty));
 
         await this._documentStorage.CreateIndexDirectoryAsync(pipeline.Index, cancellationToken).ConfigureAwait(false);
         await this._documentStorage.CreateDocumentDirectoryAsync(pipeline.Index, pipeline.DocumentId, cancellationToken).ConfigureAwait(false);

App/kernel-memory/service/Core/Pipeline/DistributedPipelineOrchestrator.cs

@@ -233,13 +233,17 @@ private async Task<bool> RunPipelineStepAsync(
             pipeline = updatedPipeline;
             pipeline.LastUpdate = DateTimeOffset.UtcNow;
 
-            this.Log.LogInformation("Handler {0} processed pipeline {1} successfully", currentStepName, pipeline.DocumentId);
+            this.Log.LogInformation("Handler {0} processed pipeline {1} successfully", 
+                currentStepName, 
+                pipeline.DocumentId.Replace("\r", string.Empty).Replace("\n", string.Empty));
             pipeline.MoveToNextStep();
             await this.MoveForwardAsync(pipeline, cancellationToken).ConfigureAwait(false);
         }
         else
         {
-            this.Log.LogError("Handler {0} failed to process pipeline {1}", currentStepName, pipeline.DocumentId);
+            this.Log.LogError("Handler {0} failed to process pipeline {1}", 
+                currentStepName, 
+                pipeline.DocumentId.Replace("\r", string.Empty).Replace("\n", string.Empty));
         }
 
         // Note: returning True, the message is removed from the queue