Skip to content

Commit d7677e5

Browse files
KSemenenkoKSemenenko
committed
Fix MEAI provider failure stream completion
1 parent 7de9b6d commit d7677e5

5 files changed

Lines changed: 190 additions & 9 deletions

File tree

‎ClaudeCodeSharpSDK.Extensions.AI/Internal/StreamingEventMapper.cs‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,12 @@ internal static async IAsyncEnumerable<ChatResponseUpdate> ToUpdates(
2020

2121
await foreach (var evt in events.WithCancellation(cancellationToken).ConfigureAwait(false))
2222
{
23+
// Keep consuming until the core process runner proves natural exit and stream EOF.
24+
if (failureMessage is not null)
25+
{
26+
continue;
27+
}
28+
2329
switch (evt)
2430
{
2531
case ThreadStartedEvent started:
@@ -87,10 +93,6 @@ internal static async IAsyncEnumerable<ChatResponseUpdate> ToUpdates(
8793
break;
8894
}
8995

90-
if (failureMessage is not null)
91-
{
92-
break;
93-
}
9496
}
9597

9698
if (failureMessage is not null)

‎ClaudeCodeSharpSDK.Tests/MEAI/ProviderFailureProcessTests.cs‎

Lines changed: 175 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,175 @@
1+
using System.Diagnostics;
2+
using ManagedCode.ClaudeCodeSharpSDK.Configuration;
3+
using ManagedCode.ClaudeCodeSharpSDK.Models;
4+
using Microsoft.Extensions.AI;
5+
6+
namespace ManagedCode.ClaudeCodeSharpSDK.Extensions.AI.Tests;
7+
8+
public sealed class ProviderFailureProcessTests
9+
{
10+
private const string ScriptFileName = "claude-provider-failure.js";
11+
private const string SandboxDirectoryName = ".sandbox";
12+
private const string TestsDirectoryName = "tests";
13+
private const string VersionFlag = "--version";
14+
private const string CliVersion = "2.1.292";
15+
private const string MarkerEnvironmentName = "CLAUDE_FAILURE_MARKER";
16+
private const string ProviderFailureLine = "{\"type\":\"result\",\"is_error\":true,\"result\":\"provider failed\"}";
17+
private const string StreamErrorLine = "{\"type\":\"error\",\"error\":\"stream failed\"}";
18+
private const string LateEventLine = "{\"type\":\"system\",\"subtype\":\"hook_started\"}";
19+
private const string UserPrompt = "bounded SDK fixture request";
20+
private const string CleanupSkipReason = "The detached inherited-pipe fixture requires POSIX process semantics.";
21+
22+
[Test]
23+
public async Task ClaudeChatClient_DrainsDelayedEventsBeforeConfirmingProviderFailureAsync()
24+
{
25+
var directory = CreateSandboxDirectory();
26+
var scriptPath = Path.Combine(directory, ScriptFileName);
27+
var markerPath = Path.Combine(directory, "natural-eof.marker");
28+
await File.WriteAllTextAsync(scriptPath, CreateDelayedExitScript());
29+
30+
try
31+
{
32+
using var client = CreateClient(scriptPath, markerPath);
33+
var updates = new List<ChatResponseUpdate>();
34+
var exception = await Assert.That(() => ConsumeAsync(client, CancellationToken.None, updates.Add)).ThrowsException();
35+
36+
await Assert.That(exception).IsTypeOf<CliExecutionFailureException>();
37+
await Assert.That(((CliExecutionFailureException)exception!).RootProcessExitConfirmed).IsTrue();
38+
await Assert.That(((CliExecutionFailureException)exception!).ExitCode).IsNull();
39+
await Assert.That(File.Exists(markerPath)).IsTrue();
40+
await Assert.That(updates.Any(static update => update.FinishReason is not null)).IsFalse();
41+
}
42+
finally
43+
{
44+
Directory.Delete(directory, recursive: true);
45+
}
46+
}
47+
48+
[Test]
49+
public async Task ClaudeChatClient_RetainedPipeCancellationDoesNotConfirmProviderFailureAsync()
50+
{
51+
if (OperatingSystem.IsWindows())
52+
{
53+
Skip.Test(CleanupSkipReason);
54+
return;
55+
}
56+
57+
var directory = CreateSandboxDirectory();
58+
var scriptPath = Path.Combine(directory, ScriptFileName);
59+
var childPidPath = Path.Combine(directory, "pipe-holder.pid");
60+
var readyPath = Path.Combine(directory, "pipe-holder.ready");
61+
await File.WriteAllTextAsync(scriptPath, CreateRetainedPipeScript(childPidPath));
62+
using var client = CreateClient(scriptPath, readyPath);
63+
using var cancellation = new CancellationTokenSource();
64+
var execution = CaptureFailureAsync(ConsumeAsync(client, cancellation.Token));
65+
66+
try
67+
{
68+
await WaitForFileAsync(readyPath);
69+
cancellation.Cancel();
70+
var exception = await execution.WaitAsync(TimeSpan.FromSeconds(12));
71+
await Assert.That(exception).IsNotNull();
72+
await Assert.That(exception).IsNotTypeOf<CliExecutionFailureException>();
73+
}
74+
finally
75+
{
76+
cancellation.Cancel();
77+
await StopChildIfPresentAsync(childPidPath);
78+
await execution.WaitAsync(TimeSpan.FromSeconds(12));
79+
Directory.Delete(directory, recursive: true);
80+
}
81+
}
82+
83+
private static ClaudeChatClient CreateClient(string scriptPath, string markerPath) => new(new ClaudeChatClientOptions
84+
{
85+
ClaudeOptions = new ClaudeOptions
86+
{
87+
ClaudeExecutablePath = scriptPath,
88+
EnvironmentVariables = new Dictionary<string, string> { [MarkerEnvironmentName] = markerPath },
89+
InheritEnvironmentVariables = true,
90+
CliMetadataProbeTimeout = TimeSpan.FromSeconds(5),
91+
ProcessTerminationTimeout = TimeSpan.FromSeconds(3),
92+
},
93+
});
94+
95+
private static async Task ConsumeAsync(
96+
ClaudeChatClient client,
97+
CancellationToken cancellationToken,
98+
Action<ChatResponseUpdate>? onUpdate = null)
99+
{
100+
await foreach (var update in client.GetStreamingResponseAsync(
101+
[new ChatMessage(ChatRole.User, UserPrompt)], cancellationToken: cancellationToken))
102+
{
103+
onUpdate?.Invoke(update);
104+
}
105+
}
106+
107+
private static async Task<Exception?> CaptureFailureAsync(Task task)
108+
{
109+
try
110+
{
111+
await task;
112+
return null;
113+
}
114+
catch (Exception exception)
115+
{
116+
return exception;
117+
}
118+
}
119+
120+
private static string CreateDelayedExitScript() => string.Concat(
121+
"const fs=require('node:fs');\n",
122+
"if(process.argv.includes('", VersionFlag, "')){console.log('", CliVersion, "');process.exit(0);}\n",
123+
"process.stdin.resume();process.stdin.on('end',()=>{process.stdout.write('", ProviderFailureLine,
124+
"\\n');setTimeout(()=>{process.stdout.write('", LateEventLine,
125+
"\\n',()=>{fs.writeFileSync(process.env.", MarkerEnvironmentName, ",'natural-eof');process.exit(0);});},150);});\n");
126+
127+
private static string CreateRetainedPipeScript(string pidPath) => string.Concat(
128+
"const fs=require('node:fs');const {spawn}=require('node:child_process');\n",
129+
"if(process.argv.includes('", VersionFlag, "')){console.log('", CliVersion, "');process.exit(0);}\n",
130+
"process.stdin.resume();process.stdin.on('end',()=>{process.stdout.write('", StreamErrorLine,
131+
"\\n',()=>setTimeout(()=>{const helperSource=\"const fs=require('node:fs');const{spawn}=require('node:child_process');const child=spawn(process.execPath,['-e','setTimeout(()=>{},30000)'],{stdio:['ignore','inherit','inherit'],detached:true});child.unref();fs.writeFileSync('",
132+
pidPath.Replace("'", "\\'", StringComparison.Ordinal),
133+
"',String(child.pid));\";const helper=spawn(process.execPath,['-e',helperSource],{stdio:['ignore','inherit','inherit'],detached:true});helper.unref();helper.on('exit',()=>fs.writeFileSync(process.env.",
134+
MarkerEnvironmentName,
135+
",'ready'));setInterval(()=>{},1000);},100));});\n");
136+
137+
private static string CreateSandboxDirectory()
138+
{
139+
var directory = Path.Combine(Environment.CurrentDirectory, TestsDirectoryName, SandboxDirectoryName,
140+
$"{Guid.NewGuid():N}");
141+
Directory.CreateDirectory(directory);
142+
return directory;
143+
}
144+
145+
private static async Task WaitForFileAsync(string path)
146+
{
147+
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10));
148+
while (!File.Exists(path))
149+
{
150+
await Task.Delay(TimeSpan.FromMilliseconds(20), timeout.Token);
151+
}
152+
}
153+
154+
private static async Task StopChildIfPresentAsync(string pidPath)
155+
{
156+
if (!File.Exists(pidPath) || !int.TryParse(await File.ReadAllTextAsync(pidPath), out var processId))
157+
{
158+
return;
159+
}
160+
161+
try
162+
{
163+
using var child = Process.GetProcessById(processId);
164+
if (!child.HasExited)
165+
{
166+
child.Kill(entireProcessTree: true);
167+
await child.WaitForExitAsync().WaitAsync(TimeSpan.FromSeconds(5));
168+
}
169+
}
170+
catch (ArgumentException)
171+
{
172+
// The fixture child already exited before cleanup.
173+
}
174+
}
175+
}

‎ClaudeCodeSharpSDK.Tests/MEAI/StreamingEventMapperTests.cs‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -108,16 +108,18 @@ public async Task ToUpdates_ClaudeToolResult_EmitsOnlySafeNativeActivityMetadata
108108
}
109109

110110
[Test]
111-
public async Task ToUpdates_TurnFailed_DisposesSourceBeforeThrowingConfirmedFailure()
111+
public async Task ToUpdates_TurnFailed_DrainsSourceBeforeThrowingConfirmedFailure()
112112
{
113113
var sourceDisposed = false;
114+
var tailConsumed = false;
114115
var exception = await Assert.That(async () =>
115-
await CollectUpdates(ProviderFailureEvents(() => sourceDisposed = true)))
116+
await CollectUpdates(ProviderFailureEvents(() => tailConsumed = true, () => sourceDisposed = true)))
116117
.ThrowsException();
117118

118119
await Assert.That(exception).IsTypeOf<CliExecutionFailureException>();
119120
await Assert.That(((CliExecutionFailureException)exception!).RootProcessExitConfirmed).IsTrue();
120121
await Assert.That(((CliExecutionFailureException)exception!).ExitCode).IsNull();
122+
await Assert.That(tailConsumed).IsTrue();
121123
await Assert.That(sourceDisposed).IsTrue();
122124
await Assert.That(exception!.Message).Contains(AuthenticationFailedMessage);
123125
}
@@ -160,11 +162,13 @@ [new ChatMessage(ChatRole.User, UserPrompt)]))
160162
}
161163
}
162164

163-
private static async IAsyncEnumerable<ThreadEvent> ProviderFailureEvents(Action onDispose)
165+
private static async IAsyncEnumerable<ThreadEvent> ProviderFailureEvents(Action onTailConsumed, Action onDispose)
164166
{
165167
try
166168
{
167169
yield return new TurnFailedEvent(new ThreadError(AuthenticationFailedMessage));
170+
yield return new ThreadErrorEvent("late stream event");
171+
onTailConsumed();
168172
await Task.Yield();
169173
}
170174
finally

‎Directory.Build.props‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@
1313
<EnforceCodeStyleInBuild>true</EnforceCodeStyleInBuild>
1414
<AnalysisLevel>latest-recommended</AnalysisLevel>
1515
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
16-
<Version>1.1.9</Version>
16+
<Version>1.1.10</Version>
1717
<PackageVersion>$(Version)</PackageVersion>
1818

1919
</PropertyGroup>

‎docs/Features/meai-integration.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -165,7 +165,7 @@ flowchart LR
165165

166166
## Definition of Done
167167

168-
Native provider failures end the mapped stream only after the upstream CLI event iterator is disposed successfully. A resulting `CliExecutionFailureException` (an `InvalidOperationException` subtype) exposes the CLI exit code when present and sets `RootProcessExitConfirmed` only after root-process exit plus natural redirected-stream EOF; this does not attest to detached descendants or external effects. If iterator cleanup is uncertain, its ordinary cleanup exception takes precedence and no confirmed-failure marker is emitted.
168+
Native provider failures stop producing response updates, but the adapter continues consuming and discarding later native events until the core CLI iterator completes naturally. Only then does it emit `CliExecutionFailureException` (an `InvalidOperationException` subtype), exposing the CLI exit code when present. `RootProcessExitConfirmed` means the SDK confirmed root-process exit and natural redirected-stream EOF; it does not attest to detached descendants or external effects. Cancellation or uncertain iterator cleanup propagates instead, so no confirmed-failure marker is emitted.
169169

170170
- `ClaudeChatClient` implements `IChatClient` with mapper coverage for the currently supported text-first adapter surface.
171171
- DI extensions register the client correctly.

0 commit comments

Comments
 (0)