Show periodic progress for import and bulk ingestion

CHANGELOG.md

@@ -11,10 +11,22 @@
 ### Fixes
 
 - Preserve `?` in unquoted command arguments so scalar indexing paths such as `index add /city/?` and `index remove /city/?` reach Cosmos DB unchanged. ([#260](https://github.com/Azure/CosmosDBShell/issues/260))
+- Long-running commands and scripts against a local emulator no longer fail with a timeout and disconnect after 10 seconds while the emulator is still responding. The 10-second limit now applies only when nothing makes progress, and it is paused while confirmation prompts and external editors wait for input.
+- Preserve partial bulk write counts through pipelines on emulator timeouts, report consumed inactivity cancellations as failures, and keep active bulk selection, export, and piped removal scans from timing out during local processing.
+- Prevent scripts from rendering timed-out partial results as successful, pause emulator inactivity deadlines during file/theme editing and change-feed polling intervals, and cancel stalled change-feed requests.
+- Keep Ctrl+C effective for startup commands and redirected scripts, and exclude local external tools such as `jq` from emulator inactivity deadlines.
+- Report interrupted startup commands as failures, retain structured partial write results, and prevent MCP prompt refreshes from canceling queued startup work.
+- Retain settled import counts and RUs on cancellation, prevent interrupted startup scripts from writing partial success output, and keep intentional watch stops successful through pipelines, blocks, and scripts.
+- Stop follow-up startup commands after an intentional watch stop in redirected stdin, and align timed-out import result error messages and exit codes with the command failure.
+- Keep import progress counts and observed RUs current when writes finish while the next input record is still being read.
+- Keep bulk completion counts current during slow input reads, allow cancellation while loading resume journals, and recognize active local validation, removal scans, and successful journal recording without treating failed requests as activity.
+- Stop interrupted startup stdin reads, confirmation prompts, and watch expressions before subsequent work; keep machine-readable stdout clean on Ctrl+C and exclude external command expressions from emulator inactivity deadlines.
 - Database and container creation on key, emulator, and static-token connections now report `created: false` and `outcome: "existing"` when an existing resource is reused, instead of always reporting `created: true`. Human-readable output says "Using existing database/container" in that case. ([#261](https://github.com/Azure/CosmosDBShell/issues/261))
 
 ### Improvements
 
+- Show periodic progress for `import`, `bulk`, and pattern-based `rm`, including elapsed time, counts, and observed RU usage, even while requests are waiting. In a terminal, a single progress line is updated in place and removed when the command finishes. Structured output and MCP responses remain unchanged.
+
 ## 1.1.291 — 2026-10-26
 
 ### Breaking changes

CosmosDBShell.Tests/CommandTests/ConnectCommandTests.cs

@@ -36,6 +36,23 @@ public void CreateClientOptions_Emulator_UsesShortRequestTimeout()
         Assert.Equal(TimeSpan.FromSeconds(5), options.RequestTimeout);
     }
 
+    [Fact]
+    public void LocalEmulatorInactivityTimeout_IsInitializedOnSharedAndNewInstances()
+    {
+        using var shell = ShellInterpreter.CreateInstance();
+
+        Assert.Equal(TimeSpan.FromSeconds(10), ShellInterpreter.Instance.LocalEmulatorInactivityTimeout);
+        Assert.Equal(TimeSpan.FromSeconds(10), shell.LocalEmulatorInactivityTimeout);
+    }
+
+    [Fact]
+    public void ReportOperationProgress_WithoutRunningCommand_IsNoOp()
+    {
+        using var shell = ShellInterpreter.CreateInstance();
+
+        shell.ReportOperationProgress();
+    }
+
     [Fact]
     public void ConnectivityFailure_LocalEmulator_DisconnectsShell()
     {

CosmosDBShell.Tests/CommandTests/ExportCommandTests.cs

@@ -122,6 +122,54 @@ public async Task WriteCsvAsync_WritesHeaderUnionAndRows()
         Assert.Equal("\"2\",\"\",\"Seattle\"", lines[2]);
     }
 
+    [Fact]
+    public async Task WriteCsvAsync_ReportsEachRowOfBothPasses()
+    {
+        var items = ToAsyncEnumerableAsync(
+            JsonSerializer.SerializeToElement(new { id = "1" }),
+            JsonSerializer.SerializeToElement(new { id = "2" }),
+            JsonSerializer.SerializeToElement(new { id = "3" }));
+        using var writer = new StringWriter();
+        var rows = 0;
+
+        var count = await ExportCommand.WriteCsvAsync(items, writer, ',', CancellationToken.None, onRowWritten: () => rows++);
+
+        Assert.Equal(3, count);
+        Assert.Equal(6, rows);
+    }
+
+    [Theory]
+    [InlineData((int)ExportFormat.Json, 1)]
+    [InlineData((int)ExportFormat.Csv, 2)]
+    public async Task WriteFileAsync_ReportsActivityDuringEveryPass(int format, int passes)
+    {
+        var directory = Directory.CreateTempSubdirectory("cosmos-export-progress-");
+        var rows = 0;
+        try
+        {
+            var count = await ExportCommand.WriteFileAsync(
+                ItemsAsync(), (ExportFormat)format, Path.Join(directory.FullName, "items"), false,
+                TestContext.Current.CancellationToken, () => rows++);
+
+            Assert.Equal(3, count);
+            Assert.Equal(3 * passes, rows);
+        }
+        finally
+        {
+            directory.Delete(recursive: true);
+        }
+
+        async IAsyncEnumerable<JsonElement> ItemsAsync()
+        {
+            for (var index = 0; index < 3; index++)
+            {
+                Assert.Equal(index, rows);
+                yield return JsonSerializer.SerializeToElement(index);
+                await Task.Yield();
+            }
+        }
+    }
+
     [Fact]
     public async Task WriteCsvAsync_EscapesSeparatorsQuotesAndNewlines()
     {

CosmosDBShell.Tests/CommandTests/ImportConcurrencyTests.cs

@@ -16,6 +16,147 @@ namespace CosmosShell.Tests.CommandTests;
 
 public class ImportConcurrencyTests
 {
+    [Theory]
+    [InlineData(0)]
+    [InlineData(1)]
+    [InlineData(2)]
+    public async Task SlowInput_RecordsSettledWritesAndSettlesReadBeforeDisposal(int failure)
+    {
+        using var timeout = CreateTimeout();
+        var container = Substitute.For<Container>();
+        var write = new TaskCompletionSource<ItemResponse<JsonElement>>(TaskCreationOptions.RunContinuationsAsynchronously);
+        ConfigureWrites(container, (_, _) => write.Task);
+        var reading = NewGate();
+        var releaseRead = NewGate();
+        var recorded = NewGate();
+        var disposed = false;
+        var progress = new ImportCommand.ImportProgress();
+        async IAsyncEnumerable<(int LineNumber, JsonElement Item)> SlowSource()
+        {
+            try
+            {
+                yield return (1, JsonSerializer.SerializeToElement(new { id = "first" }));
+                reading.TrySetResult();
+                await releaseRead.Task.WaitAsync(timeout.Token);
+            }
+            finally
+            {
+                disposed = true;
+            }
+        }
+
+        var import = ImportCommand.WriteItemsAsync(
+            SlowSource(), container, ImportMode.Insert, 2, false, [], new CommandState(), timeout.Token,
+            progress: progress,
+            progressChanged: () => recorded.TrySetResult());
+        try
+        {
+            await reading.Task.WaitAsync(timeout.Token);
+            if (failure == 1)
+            {
+                write.SetException(new InvalidOperationException("write failed"));
+            }
+            else if (failure == 2)
+            {
+                write.SetCanceled(timeout.Token);
+            }
+            else
+            {
+                write.SetResult(Response(HttpStatusCode.Created, 2.5));
+                await recorded.Task.WaitAsync(timeout.Token);
+                Assert.Equal(1, progress.Success);
+                Assert.Equal(0, progress.Failed);
+                Assert.Equal(2.5, progress.Charge);
+            }
+
+            Assert.False(import.IsCompleted);
+            Assert.False(disposed);
+        }
+        finally
+        {
+            write.TrySetResult(Response(HttpStatusCode.Created, 2.5));
+            releaseRead.TrySetResult();
+        }
+
+        if (failure == 1)
+        {
+            var error = await Assert.ThrowsAsync<InvalidOperationException>(() => import.WaitAsync(timeout.Token));
+            Assert.Equal("write failed", error.Message);
+        }
+        else if (failure == 2)
+        {
+            await Assert.ThrowsAnyAsync<OperationCanceledException>(() => import.WaitAsync(timeout.Token));
+        }
+        else
+        {
+            Assert.Equal((1, 0, 2.5), await import.WaitAsync(timeout.Token));
+        }
+
+        Assert.True(disposed);
+    }
+
+    [Fact]
+    public async Task FinalDrain_RecordsWritesThatFinishBehindASlowerWrite()
+    {
+        using var timeout = CreateTimeout();
+        var container = Substitute.For<Container>();
+        var gates = new[] { NewGate(), NewGate() };
+        var started = new[] { NewGate(), NewGate() };
+        ConfigureWrites(container, async (item, token) =>
+        {
+            var index = item.GetProperty("id").GetInt32();
+            started[index].TrySetResult();
+            await gates[index].Task.WaitAsync(token);
+            return Response(HttpStatusCode.Created, 1);
+        });
+        using var reader = NewReader(2);
+        var progress = new ImportCommand.ImportProgress();
+        var secondRecorded = NewGate();
+        var import = ImportCommand.WriteItemsAsync(
+            ImportCommand.EnumerateJsonLinesAsync(reader, timeout.Token),
+            container, ImportMode.Insert, 3, false, [], new CommandState(), timeout.Token,
+            progress: progress,
+            progressChanged: () => secondRecorded.TrySetResult());
+        try
+        {
+            await Task.WhenAll(started.Select(gate => gate.Task)).WaitAsync(timeout.Token);
+            gates[1].TrySetResult();
+
+            await secondRecorded.Task.WaitAsync(timeout.Token);
+            Assert.Equal(1, progress.Success);
+            Assert.False(import.IsCompleted);
+        }
+        finally
+        {
+            gates[0].TrySetResult();
+            gates[1].TrySetResult();
+        }
+
+        Assert.Equal((2, 0, 2.0), await import.WaitAsync(timeout.Token));
+    }
+
+    [Fact]
+    public async Task ProgressUpdatesIncludeEverySettledWriteAndObservedCharge()
+    {
+        using var timeout = CreateTimeout();
+        var container = Substitute.For<Container>();
+        ConfigureWrites(container, (item, _) => Task.FromResult(
+            Response(item.GetProperty("id").GetInt32() == 1 ? HttpStatusCode.Conflict : HttpStatusCode.Created, 2.5)));
+        using var reader = NewReader(3);
+        var progress = new ImportCommand.ImportProgress();
+        var snapshots = new List<(int Success, int Failed, double Charge)>();
+
+        await ImportCommand.WriteItemsAsync(
+            ImportCommand.EnumerateJsonLinesAsync(reader, timeout.Token),
+            container, ImportMode.Insert, 3, true, [], new CommandState(), timeout.Token,
+            progress: progress,
+            progressChanged: () => snapshots.Add((progress.Success, progress.Failed, progress.Charge)));
+
+        Assert.Equal(3, snapshots.Count);
+        Assert.Equal((2, 1, 7.5), snapshots[^1]);
+        Assert.Equal([1, 2, 3], snapshots.Select(snapshot => snapshot.Success + snapshot.Failed));
+    }
+
     [Theory]
     [InlineData(1)]
     [InlineData(3)]

CosmosDBShell.Tests/CommandTests/ImportFailureResultTests.cs

@@ -15,10 +15,133 @@ namespace CosmosShell.Tests.CommandTests;
 using Azure.Data.Cosmos.Shell.Util;
 using Microsoft.Azure.Cosmos;
 using NSubstitute;
+using CosmosShell.Tests.Integration;
 
 [Collection(ConsoleOutputTestCollection.Name)]
 public class ImportFailureResultTests
 {
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task EmulatorTimeout_SynchronizesStructuredClassificationAndPreservesSettledWrites(bool block)
+    {
+        var file = Path.GetTempFileName();
+        var errors = Path.GetTempFileName();
+        var output = Path.GetTempFileName();
+        using var shell = CreateShell(out var container);
+        Assert.IsAssignableFrom<ConnectedState>(shell.State).Client!.Endpoint.Returns(new Uri("https://localhost:8081"));
+        shell.Options = new Program.CosmosShellOptions { Output = "json", Quiet = true };
+        shell.LocalEmulatorInactivityTimeout = TimeSpan.FromMilliseconds(100);
+        shell.ErrOutRedirect = errors;
+        shell.StdOutRedirect = output;
+        var calls = 0;
+        container.CreateItemStreamAsync(Arg.Any<Stream>(), Arg.Any<PartitionKey>(), Arg.Any<ItemRequestOptions>(), Arg.Any<CancellationToken>())
+            .Returns(async call =>
+            {
+                if (++calls > 1)
+                {
+                    await Task.Delay(Timeout.InfiniteTimeSpan, call.Arg<CancellationToken>());
+                }
+
+                return Response(2);
+            });
+        try
+        {
+            await File.WriteAllTextAsync(file, """[{"id":"first","pk":"x"},{"id":"second","pk":"x"}]""", TestContext.Current.CancellationToken);
+            var command = $"import {JsonSerializer.Serialize(file)} --concurrency=1";
+            if (block)
+            {
+                command = $"{{ {command}; echo downstream }}";
+            }
+
+            var state = Assert.IsType<StructuredErrorCommandState>(await shell.ExecuteCommandAsync(command, TestContext.Current.CancellationToken));
+
+            Assert.Equal(ShellExitCode.ConnectionError, state.ExitCode);
+            Assert.IsType<DisconnectedState>(shell.State);
+            var result = Assert.IsType<ShellJson>(state.Result).Value;
+            Assert.Equal(ShellExitCode.ConnectionError, result.GetProperty("exitCode").GetInt32());
+            Assert.Equal(state.Exception.Message, result.GetProperty("error").GetString());
+            Assert.Contains("timed out", state.Exception.Message, StringComparison.OrdinalIgnoreCase);
+            Assert.Equal(state.Exception.Message, IntegrationTestBase.GetErrorMessage(state));
+            Assert.Equal(1, result.GetProperty("imported").GetInt32());
+            Assert.Equal(0, result.GetProperty("failed").GetInt32());
+            Assert.Equal(2, result.GetProperty("requestCharge").GetDouble());
+            Assert.Equal(3.25, state.RequestCharge);
+            Assert.Empty(await File.ReadAllTextAsync(output, TestContext.Current.CancellationToken));
+            using var error = JsonDocument.Parse(await File.ReadAllTextAsync(errors, TestContext.Current.CancellationToken));
+            Assert.True(JsonElement.DeepEquals(result, error.RootElement.GetProperty("result")));
+            var mcp = McpResponseFactory.CreateSuccess(state, shell.State);
+            Assert.True(mcp.IsError);
+            Assert.True(JsonElement.DeepEquals(result, Assert.IsType<JsonElement>(mcp.StructuredContent).GetProperty("result")));
+        }
+        finally
+        {
+            shell.StdOutRedirect = null;
+            shell.ErrOutRedirect = null;
+            File.Delete(file);
+            File.Delete(errors);
+            File.Delete(output);
+        }
+    }
+
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task StartupCancellation_PreservesSettledWritesAndRendersPartialResult(bool cancelInFlight)
+    {
+        var file = Path.GetTempFileName();
+        var errors = Path.GetTempFileName();
+        var output = Path.GetTempFileName();
+        using var shell = CreateShell(out var container);
+        shell.Options = new Program.CosmosShellOptions { Output = "json", Quiet = true };
+        shell.ErrOutRedirect = errors;
+        shell.StdOutRedirect = output;
+        var calls = 0;
+        container.CreateItemStreamAsync(Arg.Any<Stream>(), Arg.Any<PartitionKey>(), Arg.Any<ItemRequestOptions>(), Arg.Any<CancellationToken>())
+            .Returns(call =>
+            {
+                calls++;
+                if (calls == (cancelInFlight ? 2 : 1))
+                {
+                    shell.CancelPrompt();
+                    if (cancelInFlight)
+                    {
+                        return Task.FromCanceled<ResponseMessage>(call.Arg<CancellationToken>());
+                    }
+                }
+
+                return Task.FromResult(Response(2));
+            });
+        try
+        {
+            await File.WriteAllTextAsync(file, """[{"id":"first","pk":"x"},{"id":"second","pk":"x"},{"id":"third","pk":"x"}]""", TestContext.Current.CancellationToken);
+
+            var (state, interrupted) = await Program.ExecuteStartupCommandAsync(shell, $"import {JsonSerializer.Serialize(file)} --concurrency=1");
+
+            Assert.IsType<StructuredErrorCommandState>(state);
+            Assert.True(interrupted);
+            Assert.True(state.IsError);
+            var result = Assert.IsType<ShellJson>(state.Result).Value;
+            Assert.Equal(1, result.GetProperty("imported").GetInt32());
+            Assert.Equal(0, result.GetProperty("failed").GetInt32());
+            Assert.Equal(2, result.GetProperty("requestCharge").GetDouble());
+            Assert.Equal(3.25, state.RequestCharge);
+            Assert.Equal(cancelInFlight ? 2 : 1, calls);
+            Assert.Empty(await File.ReadAllTextAsync(output, TestContext.Current.CancellationToken));
+            using var error = JsonDocument.Parse(await File.ReadAllTextAsync(errors, TestContext.Current.CancellationToken));
+            Assert.Equal("error", error.RootElement.GetProperty("status").GetString());
+            Assert.True(JsonElement.DeepEquals(result, error.RootElement.GetProperty("result")));
+        }
+        finally
+        {
+            shell.StdOutRedirect = null;
+            shell.ErrOutRedirect = null;
+            File.Delete(file);
+            File.Delete(errors);
+            File.Delete(output);
+        }
+    }
+
     [Theory]
     [InlineData(429, ShellExitCode.Throttled, false, false)]
     [InlineData(429, ShellExitCode.Throttled, true, false)]