Make a cancelled turn actually unwind in the Go and .NET servers

.changeset/cancel-unwinds-the-turn.md

@@ -0,0 +1,13 @@
+---
+'@smooai/smooth-operator': patch
+---
+
+go-server / dotnet-server: make cancelling a turn a **stop** button rather than a **mute** button.
+
+After a client cancels, the Go and .NET servers walked away from the turn — the runner returns at its first cancellation check and the connection's sink is gagged — but the **agent loop kept running**. The engine folds every tool failure back to the model as a tool result and iterates, and after a cancel that failure is the `context canceled` a tool returns, or the denial the write-confirmation gate returns once `TryCancelActiveTurn` unparks it. So the loop went on to another model call and acted on the answer, with every trace of it discarded. Rust, TypeScript and Python unwind properly; this brings the two ports in line.
+
+Neither engine's loop has a cancellation check of its own, and cancellation in Go/.NET is cooperative rather than the preemptive future-drop the Rust reference gets for free. The loop is therefore stopped at the one place it re-enters shared state — the model call: the turn's chat client is wrapped so a cancelled context fails the call instead of issuing it, which unwinds `RunStream` / `RunStreamingAsync` and ends the turn. In production the gateway client would have failed that call on its own cancelled context; the servers now stop the turn themselves instead of relying on the transport to do it.
+
+This also clears the `DATA RACE` the Go race detector reports on the shared conformance corpus's `cancel-mid-turn` scenario, where the cancelled turn's goroutine and the next turn's goroutine drove the engine's mock provider concurrently — independent proof that the cancelled turn was still running.
+
+Regression tests: `TestCancelledTurnMakesNoFurtherModelCall` (Go) and `ACancelledTurn_MakesNoFurtherModelCall_SoTheNextTurnKeepsItsResponse` (.NET). Both assert the model-call count at a settle point rather than on timing, and both fail without the fix.

dotnet/server/integration-tests/MockChatClient.cs

@@ -27,8 +27,14 @@ public MockChatClient PushToolCall(string callId, string name, IDictionary<strin
         return this;
     }
 
-    private ChatResponse Next() =>
-        _responses.Count > 0 ? _responses.Dequeue() : new ChatResponse(new ChatMessage(ChatRole.Assistant, string.Empty));
+    /// <summary>How many model calls the server actually made (a cancelled turn must make none).</summary>
+    public int Calls;
+
+    private ChatResponse Next()
+    {
+        Interlocked.Increment(ref Calls);
+        return _responses.Count > 0 ? _responses.Dequeue() : new ChatResponse(new ChatMessage(ChatRole.Assistant, string.Empty));
+    }
 
     public Task<ChatResponse> GetResponseAsync(IEnumerable<ChatMessage> messages, ChatOptions? options = null, CancellationToken cancellationToken = default) =>
         Task.FromResult(Next());

dotnet/server/integration-tests/TurnCancelTests.cs

@@ -66,7 +66,7 @@ private sealed class SlowToolProbe
             "parks the turn for cancellation tests");
     }
 
-    private static WebApplication BuildApp(MockChatClient chat, AITool? tool = null)
+    private static WebApplication BuildApp(MockChatClient chat, AITool? tool = null, IReadOnlyList<string>? confirmTools = null)
     {
         var builder = WebApplication.CreateBuilder();
         builder.WebHost.UseTestServer();
@@ -75,6 +75,10 @@ private static WebApplication BuildApp(MockChatClient chat, AITool? tool = null)
         {
             builder.Services.AddSingleton<IReadOnlyList<AITool>>(new[] { tool });
         }
+        if (confirmTools is { Count: > 0 })
+        {
+            builder.Services.AddSingleton(new ConfirmTools(confirmTools));
+        }
         builder.Services.AddSmoothOperatorServer();
 
         var app = builder.Build();
@@ -190,6 +194,70 @@ public async Task CancelMidTurn_AbortsTheTurn_AndEmitsCancelled()
         await app.StopAsync();
     }
 
+    /// <summary>
+    /// Cancel is a STOP button, not a mute button (th-f2ac48).
+    ///
+    /// <para>A turn parked at a write-confirmation is unparked as DENIED when it is cancelled (the park
+    /// awaits a bare task the CTS cannot complete, so <c>TryCancelActiveTurn</c> resolves it rather than
+    /// leave it hung). The engine folds that denial back to the model as a tool result and iterates —
+    /// so the cancelled turn used to call the model AGAIN, and act on the answer, while the gagged sink
+    /// hid every bit of it.</para>
+    ///
+    /// <para>Asserted on deterministic state, not timing: the model call COUNT, read once the host has
+    /// shut down. Shutdown is the settle point that makes it deterministic — a cancelled turn is
+    /// orphaned (<c>TryCancelActiveTurn</c> clears the turn slot, so <c>WaitForTurnsAsync</c> no longer
+    /// tracks it), and teardown's <c>RejectPendingConfirmations</c> is what finally unparks it. Without
+    /// the fix that unpark leads straight to a third model call.</para>
+    /// </summary>
+    [Fact]
+    public async Task ACancelledTurn_MakesNoFurtherModelCall_SoTheNextTurnKeepsItsResponse()
+    {
+        const string GatedTool = "delete_record";
+        const string SecondTurnReply = "the second turn got its own response";
+
+        // Turn 1 calls the confirm-gated tool and parks; the text is scripted for TURN 2, and a
+        // cancelled turn 1 must not consume it.
+        var chat = new MockChatClient()
+            .PushToolCall("call-1", GatedTool, new Dictionary<string, object?> { ["id"] = "42" })
+            .PushText(SecondTurnReply);
+        var tool = AIFunctionFactory.Create(() => "Record 42 deleted.", GatedTool, "Delete a record by id (a state-mutating write).");
+
+        await using var app = BuildApp(chat, tool, new[] { GatedTool });
+        await app.StartAsync();
+        using var socket = await ConnectAsync(app.GetTestServer());
+        var sessionId = await CreateSessionAsync(socket);
+
+        await SendAsync(socket, new JsonObject
+        {
+            ["action"] = "send_message", ["requestId"] = "turn-1", ["sessionId"] = sessionId, ["message"] = "delete record 42", ["stream"] = true,
+        });
+
+        // The park prompt is the proof the turn is genuinely in flight.
+        var seen = new List<JsonObject>();
+        await ReadUntilAsync(socket, "write_confirmation_required", seen);
+
+        await SendAsync(socket, new JsonObject { ["action"] = "cancel", ["requestId"] = "turn-1", ["sessionId"] = sessionId });
+        await ReadUntilAsync(socket, "cancelled", seen);
+
+        await SendAsync(socket, new JsonObject
+        {
+            ["action"] = "send_message", ["requestId"] = "turn-2", ["sessionId"] = sessionId, ["message"] = "never mind, say something", ["stream"] = true,
+        });
+        var settled = await ReadUntilAsync(socket, "eventual_response", seen);
+
+        Assert.Equal("turn-2", settled["requestId"]!.GetValue<string>());
+        var parts = settled["data"]!["data"]!["response"]!["responseParts"]!.AsArray();
+        Assert.Equal(SecondTurnReply, string.Concat(parts.Select(p => p!.GetValue<string>())));
+
+        await socket.CloseAsync(WebSocketCloseStatus.NormalClosure, "done", CancellationToken.None);
+        await app.StopAsync();
+
+        // Exactly one model call per turn: the cancelled turn made its first call and stopped. Read
+        // after the host has shut down — teardown rejects any outstanding confirmation and drains the
+        // connection, so a turn that was still going would have made its extra call by now.
+        Assert.Equal(2, chat.Calls);
+    }
+
     [Fact]
     public async Task CancelWithNoActiveTurn_IsASilentNoop()
     {

dotnet/server/src/TurnRunner.cs

@@ -379,7 +379,7 @@ public async Task<TurnResult> RunAsync(string conversationId, string requestId,
             }
         }
 
-        var agent = new SmoothAgent(_chatClient, options);
+        var agent = new SmoothAgent(new CancelAwareChatClient(_chatClient), options);
         var thread = agent.GetNewThread();
         foreach (var message in priorMessages)
         {
@@ -756,3 +756,50 @@ private static JsonObject ToolResultState(string name, FunctionResultContent res
         };
     }
 }
+
+/// <summary>
+/// Wraps the turn's chat client so a CANCELLED turn can never issue another model call — the
+/// difference between a stop button and a mute button.
+///
+/// <para>The engine's agent loop has no cancellation check of its own: it folds every tool failure
+/// back to the model as a result and iterates, including the denial the write-confirmation gate
+/// returns once the turn is cancelled (<c>TryCancelActiveTurn</c> resolves the park as denied so it
+/// cannot hang). The runner has already walked away by then and its sink is gagged, so the loop's
+/// remaining output is discarded — but the loop itself keeps running, calling the model again and
+/// acting on whatever it answers, on a turn the user stopped.</para>
+///
+/// <para>The live gateway client would fail that call on its own cancelled token, so this is not a
+/// standing spend leak — it is that the server was RELYING on the transport to stop a cancelled
+/// turn. Cancellation is cooperative here, so the loop is stopped at the one place it
+/// re-enters shared state: the model call. Throwing on a cancelled token unwinds
+/// <c>RunStreamingAsync</c> and the turn ends — the .NET analog of dropping the Rust turn future,
+/// which is preemptive and needs no such guard. The runner's caller already treats
+/// <see cref="OperationCanceledException"/> as a clean cancellation.</para>
+///
+/// <para>Per-turn and decorative only: <see cref="Dispose"/> does NOT dispose the wrapped client,
+/// which the server owns and reuses across turns.</para>
+/// </summary>
+internal sealed class CancelAwareChatClient : IChatClient
+{
+    private readonly IChatClient _inner;
+
+    public CancelAwareChatClient(IChatClient inner) => _inner = inner;
+
+    public Task<ChatResponse> GetResponseAsync(IEnumerable<ChatMessage> messages, ChatOptions? options = null, CancellationToken cancellationToken = default)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        return _inner.GetResponseAsync(messages, options, cancellationToken);
+    }
+
+    public IAsyncEnumerable<ChatResponseUpdate> GetStreamingResponseAsync(IEnumerable<ChatMessage> messages, ChatOptions? options = null, CancellationToken cancellationToken = default)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        return _inner.GetStreamingResponseAsync(messages, options, cancellationToken);
+    }
+
+    public object? GetService(Type serviceType, object? serviceKey = null) => _inner.GetService(serviceType, serviceKey);
+
+    public void Dispose()
+    {
+    }
+}

go/server/turn_cancel_test.go

@@ -101,8 +101,9 @@ func (p *slowToolProbe) awaitCancelled(t *testing.T) {
 
 // slowToolServer spins up a server whose mock scripts a single call to the parking tool
 // (so the turn parks and never returns on its own), then a text reply that a
-// non-cancelled turn would settle with.
-func slowToolServer(t *testing.T, probe *slowToolProbe) *LocalServer {
+// non-cancelled turn would settle with. The mock is returned so a test can assert how
+// many model calls the server actually made.
+func slowToolServer(t *testing.T, probe *slowToolProbe) (*LocalServer, *core.MockLlmProvider) {
 	t.Helper()
 	mock := core.NewMockLlmProvider()
 	mock.PushToolCall("call-1", cancelSlowTool, `{}`)
@@ -117,7 +118,7 @@ func slowToolServer(t *testing.T, probe *slowToolProbe) *LocalServer {
 		t.Fatalf("spawn: %v", err)
 	}
 	t.Cleanup(func() { _ = ls.Shutdown() })
-	return ls
+	return ls, mock
 }
 
 // nextRawEv reads the next server event WITHOUT the pong/keepalive filtering nextEv
@@ -191,7 +192,7 @@ func recvUntil(t *testing.T, transport protocol.Transport, typ string, within ti
 // connection stays usable.
 func TestCancelMidTurnAbortsAndEmitsCancelled(t *testing.T) {
 	probe := newSlowToolProbe()
-	ls := slowToolServer(t, probe)
+	ls, _ := slowToolServer(t, probe)
 	transport := connectTransport(t, ls)
 	defer func() { _ = transport.Close() }()
 
@@ -257,6 +258,58 @@ func TestCancelMidTurnAbortsAndEmitsCancelled(t *testing.T) {
 	}
 }
 
+// TestCancelledTurnMakesNoFurtherModelCall — cancel is a STOP button, not a mute button.
+//
+// The regression this guards (th-f2ac48): the engine's agent loop folds a tool failure —
+// including the `context canceled` a tool or the write-confirmation gate returns once the
+// turn is cancelled — back to the model as a tool result and iterates. The runner has
+// already walked away, so that next model call and everything it asks for happened on a
+// turn the user stopped, with its output merely discarded.
+//
+// Asserted on deterministic state, not timing: the mock's call count, and the fact that
+// the NEXT turn still finds its scripted response. Without the fix the cancelled turn
+// eats the second script entry, so turn 2 streams nothing (and `go test -race` reports the
+// two turns popping the mock's FIFO concurrently — the same defect from the other side).
+func TestCancelledTurnMakesNoFurtherModelCall(t *testing.T) {
+	probe := newSlowToolProbe()
+	ls, mock := slowToolServer(t, probe)
+	transport := connectTransport(t, ls)
+	defer func() { _ = transport.Close() }()
+
+	sessionID := createSession(t, transport)
+
+	sendFrame(t, transport, map[string]any{
+		"action":    "send_message",
+		"requestId": "turn-1",
+		"sessionId": sessionID,
+		"message":   "please do the slow thing",
+	})
+	probe.awaitStart(t)
+	sendFrame(t, transport, map[string]any{"action": "cancel", "requestId": "turn-1"})
+	recvUntil(t, transport, "cancelled", 5*time.Second)
+	probe.awaitCancelled(t)
+
+	// The second scripted response is still there for the NEXT turn — the cancelled turn
+	// did not consume it. This is the assertion that fails without the fix.
+	sendFrame(t, transport, map[string]any{
+		"action":    "send_message",
+		"requestId": "turn-2",
+		"sessionId": sessionID,
+		"message":   "never mind, say something",
+	})
+	done, _ := recvUntil(t, transport, "eventual_response", 10*time.Second)
+	parts, ok := dot(t, done, "data.data.response.responseParts")
+	if !ok || !jsonEqual(parts, []any{"Finished the slow thing."}) {
+		t.Fatalf("turn 2 must receive the response the cancelled turn did not consume, got %s", mustJSON(done))
+	}
+
+	// Exactly one model call per turn: the cancelled turn made its first call and stopped.
+	// Read once both turns have settled, so this is state, not a race with a live turn.
+	if got := mock.CallCount(); got != 2 {
+		t.Fatalf("model calls = %d, want 2 (one per turn); a cancelled turn must not call the model again", got)
+	}
+}
+
 // TestCancelWithNoActiveTurnIsNoop — a cancel with nothing running emits nothing and
 // leaves the connection live.
 func TestCancelWithNoActiveTurnIsNoop(t *testing.T) {
@@ -325,7 +378,7 @@ func TestNormalTurnStillCompletes(t *testing.T) {
 // (no client remains to receive its output).
 func TestDisconnectMidTurnAbortsTheTurn(t *testing.T) {
 	probe := newSlowToolProbe()
-	ls := slowToolServer(t, probe)
+	ls, _ := slowToolServer(t, probe)
 	transport := connectTransport(t, ls)
 
 	sessionID := createSession(t, transport)
@@ -355,7 +408,7 @@ func TestDisconnectMidTurnAbortsTheTurn(t *testing.T) {
 // concurrently.
 func TestSecondSendMessageWhileTurnInFlightIsRejected(t *testing.T) {
 	probe := newSlowToolProbe()
-	ls := slowToolServer(t, probe)
+	ls, _ := slowToolServer(t, probe)
 	transport := connectTransport(t, ls)
 	defer func() { _ = transport.Close() }()
 

go/server/turn_runner.go

@@ -308,7 +308,7 @@ func (r *TurnRunner) Run(ctx context.Context, sessionID, conversationID, request
 		}
 	}
 
-	agent := core.NewSmoothAgent(r.client, opts)
+	agent := core.NewSmoothAgent(cancelAware(r.client), opts)
 	thread := core.NewThread()
 	prior, err := r.store.ListMessages(ctx, conversationID, maxPriorMessages)
 	if err != nil {
@@ -525,6 +525,55 @@ consume:
 	return TurnResult{Reply: reply.String(), MessageID: outbound.ID, Citations: citations, NextStepID: nextStepID, Usage: usage}, nil
 }
 
+// cancelAware wraps the engine's chat client so a CANCELLED turn can never issue
+// another model call — the difference between a stop button and a mute button.
+//
+// The engine's agent loop has no cancellation check of its own: it folds every tool
+// failure back to the model as a tool result and iterates, including the
+// `context canceled` a tool (or the write-confirmation gate) returns once the turn is
+// cancelled. The runner has already walked away by then — it returns at the first
+// ctx.Done() and drains the stream — so the loop's remaining output is discarded, but
+// the loop itself keeps running: another model call, and whatever that call asks for,
+// on a turn the user stopped.
+//
+// The live GatewayClient would fail that call on its own cancelled context, so this is
+// not a standing spend leak — it is that the server was RELYING on the transport to
+// stop a cancelled turn. Cancellation in Go is cooperative, so the loop is stopped here
+// instead, at the one place it re-enters shared state: the model call. Failing it on a
+// cancelled context aborts runStream (`model stream: context canceled`) and the turn
+// unwinds — the Go analog of dropping the Rust turn future, which is preemptive and
+// needs no such guard.
+//
+// A client that does not stream is returned unchanged, so the engine's
+// StreamingChatClient assertion still fails exactly as it did before.
+func cancelAware(client core.ChatClient) core.ChatClient {
+	streaming, ok := client.(core.StreamingChatClient)
+	if !ok {
+		return client
+	}
+	return cancelAwareClient{inner: streaming}
+}
+
+// cancelAwareClient is the wrapper cancelAware installs: every model call checks the
+// turn's context first and fails rather than reaching the gateway.
+type cancelAwareClient struct {
+	inner core.StreamingChatClient
+}
+
+func (c cancelAwareClient) Chat(ctx context.Context, req core.ChatRequest) (core.ChatResponse, error) {
+	if err := ctx.Err(); err != nil {
+		return core.ChatResponse{}, err
+	}
+	return c.inner.Chat(ctx, req)
+}
+
+func (c cancelAwareClient) ChatStream(ctx context.Context, req core.ChatRequest) (<-chan core.ChatChunk, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	return c.inner.ChatStream(ctx, req)
+}
+
 // drainStream discards the tail of an abandoned engine stream so its producer
 // goroutine — blocked mid-send on an unbuffered channel — can finish and close it.
 // Only reached when a turn is cancelled; the engine shares the cancelled context, so