Skip to content

Add file progress updates to pipeline #11

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
elbruno merged 1 commit into from
Jul 1, 2026
Merged

Add file progress updates to pipeline #11

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 36 additions & 1 deletion src/Graphify.Cli/PipelineRunner.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ public sealed class PipelineRunner
private readonly TextWriter _output;
private readonly bool _verbose;
private readonly IChatClient? _chatClient;
private readonly SemaphoreSlim _outputLock = new(1, 1);

public PipelineRunner(TextWriter output, bool verbose = false, IChatClient? chatClient = null)
{
Expand Down Expand Up @@ -66,7 +67,9 @@ public PipelineRunner(TextWriter output, bool verbose = false, IChatClient? chat
var extractionBag = new ConcurrentBag<ExtractionResult>();
int processed = 0;
int skipped = 0;
int completed = 0;
var verboseWarnings = new ConcurrentQueue<string>();
var extractionProgressStep = Math.Max(1, detectedFiles.Count / 10);

await Parallel.ForEachAsync(
detectedFiles,
Expand All @@ -89,10 +92,21 @@ await Parallel.ForEachAsync(
{
Interlocked.Increment(ref skipped);
}

var finished = Interlocked.Increment(ref completed);
if (ShouldReportProgress(finished, detectedFiles.Count, extractionProgressStep))
{
await WriteLineAsync($" Progress: {finished}/{detectedFiles.Count} files analyzed");
}
}
catch (Exception ex)
{
Interlocked.Increment(ref skipped);
var finished = Interlocked.Increment(ref completed);
if (ShouldReportProgress(finished, detectedFiles.Count, extractionProgressStep))
{
await WriteLineAsync($" Progress: {finished}/{detectedFiles.Count} files analyzed");
}
if (_verbose)
{
verboseWarnings.Enqueue($" Warning: Failed to extract {file.RelativePath}: {ex.Message}");
Expand All @@ -118,6 +132,8 @@ await Parallel.ForEachAsync(
await WriteLineAsync("[2b/6] Running AI-enhanced semantic extraction...");
var semanticExtractor = new SemanticExtractor(_chatClient);
int semanticProcessed = 0;
int semanticCompleted = 0;
var semanticProgressStep = Math.Max(1, detectedFiles.Count / 10);

foreach (var file in detectedFiles)
{
Expand All @@ -136,6 +152,12 @@ await Parallel.ForEachAsync(
if (_verbose)
await WriteLineAsync($" Warning: Semantic extraction failed for {file.RelativePath}: {ex.Message}");
}

var finished = ++semanticCompleted;
if (ShouldReportProgress(finished, detectedFiles.Count, semanticProgressStep))
{
await WriteLineAsync($" Progress: {finished}/{detectedFiles.Count} files semantically analyzed");
}
}

await WriteLineAsync($" AI extracted from {semanticProcessed} files");
Expand Down Expand Up @@ -333,7 +355,20 @@ await Parallel.ForEachAsync(

private async Task WriteLineAsync(string message = "")
{
await _output.WriteLineAsync(message);
await _outputLock.WaitAsync();
try
{
await _output.WriteLineAsync(message);
}
finally
{
_outputLock.Release();
}
}

private static bool ShouldReportProgress(int completed, int total, int step)
{
return completed == total || completed % step == 0;
}

private static Dictionary<int, string> BuildCommunityLabels(KnowledgeGraph graph)
Expand Down
8 changes: 8 additions & 0 deletions src/Graphify/Pipeline/WatchMode.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,6 +188,8 @@ private async Task ProcessChangesAsync(
// Extract
var extractor = new Extractor();
var newResults = new List<ExtractionResult>();
var progressStep = Math.Max(1, filesToProcess.Count / 10);
var completed = 0;
foreach (var file in filesToProcess)
{
ct.ThrowIfCancellationRequested();
Expand All @@ -202,6 +204,12 @@ private async Task ProcessChangesAsync(
if (_verbose)
await _output.WriteLineAsync($" Warning: extraction failed for {file.RelativePath}: {ex.Message}");
}

completed++;
if (completed == filesToProcess.Count || completed % progressStep == 0)
{
await _output.WriteLineAsync($" Progress: {completed}/{filesToProcess.Count} changed files analyzed");
}
}

if (newResults.Count == 0)
Expand Down
30 changes: 30 additions & 0 deletions src/tests/Graphify.Tests/Cli/PipelineRunnerTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,4 +72,34 @@ public async Task RunAsync_LadybugFormat_RoutesToLadybugExporter()
try { if (Directory.Exists(tempOutput)) Directory.Delete(tempOutput, recursive: true); } catch { /* best-effort cleanup */ }
}
}

[Fact]
[Trait("Category", "Cli")]
public async Task RunAsync_ReportsFileProgressDuringExtraction()
{
var tempInput = Path.Combine(Path.GetTempPath(), Guid.NewGuid().ToString("N"));
var tempOutput = Path.Combine(Path.GetTempPath(), Guid.NewGuid().ToString("N"));
Directory.CreateDirectory(tempInput);

try
{
await File.WriteAllTextAsync(Path.Combine(tempInput, "alpha.cs"), "namespace Demo; public sealed class Alpha { }");
await File.WriteAllTextAsync(Path.Combine(tempInput, "beta.cs"), "namespace Demo; public sealed class Beta { }");

var sb = new StringBuilder();
await using var writer = new StringWriter(sb);
var runner = new PipelineRunner(writer, verbose: false);

await runner.RunAsync(tempInput, tempOutput, formats: ["json"], useCache: false);

var log = sb.ToString();
Assert.Contains("Progress:", log, StringComparison.OrdinalIgnoreCase);
Assert.Contains("2/2", log, StringComparison.OrdinalIgnoreCase);
}
finally
{
try { Directory.Delete(tempInput, recursive: true); } catch { /* best-effort cleanup */ }
try { if (Directory.Exists(tempOutput)) Directory.Delete(tempOutput, recursive: true); } catch { /* best-effort cleanup */ }
}
}
}
Loading