fix: v1 prompt cancellation targets its own turn, answers after the session's cancel, one turn at a time

AGENTS.md

@@ -149,6 +149,7 @@ Lifecycle correctness is part of the public behavior. Changes to startup, cancel
 - Return follow-up work and cleanup through `RequestOutcome` rather than running them inside the handler.
 - Keep normal close, timeout/abort, parent cancellation, and I/O failure paths distinct and testable.
 - The v1 client keeps one `ClientSessionHolder` per session id. With a `ClientOperationsFactory`, the holder buffers the session's updates until the opening call answers. The `loadSession` and `resumeSession` overloads that take `ClientSessionOperations` claim the holder before the request goes out, and one consumer hands its queue to the operations until the answer. Until the switch to the session, every update passes that one queue with its one consumer, so the arrival order holds by construction: do not add a second delivery path, a lock, or a counter there. After the switch, an update goes synchronously to the session. An update that misses the closed queue waits in the `lateDeliveries` chain for the drain and for the update that missed the queue before it, so late updates keep their order too.
+- The v1 agent runs one turn at a time per session, an SDK guarantee the protocol does not define. Only the turn frees its slot, when it answers, so a `session/prompt` sent earlier, even while a cancelled turn winds down, is rejected. `session/cancel` targets the turn registered when it arrives, never one registered while `AgentSession.cancel()` runs, and a cancelled turn answers only after every such call returns, so updates sent from it precede the response.
 
 For WebSocket shutdown, preserve the established sequence: reject new work, enter closing state, drain accepted frames within the configured timeout, then perform a normal WebSocket close. Reserve cancellation/abort for timeout, cancellation, or writer failure.
 

acp/src/commonMain/kotlin/com/agentclientprotocol/agent/Agent.kt

@@ -17,6 +17,10 @@ import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.getAndUpdate
+import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
 import kotlinx.serialization.json.JsonElement
@@ -62,59 +66,119 @@ public class Agent(
         val clientOperations: ClientSessionOperations,
         protocol: Protocol
     ) : BaseSessionWrapper(agent, protocol) {
-        private class PromptSession(val promptJob: Job)
+        private class PromptSession(val promptJob: Job) {
+            /**
+             * The number of cancels still running for this turn, or [ANSWERING] once the turn has passed the wait for
+             * them and is about to answer.
+             */
+            private val runningCancels = MutableStateFlow(0)
+
+            private val _isCancelRequested = atomic(false)
+
+            /**
+             * Whether a `session/cancel` targeted this turn before it started answering.
+             */
+            val isCancelRequested: Boolean get() = _isCancelRequested.value
+
+            /**
+             * Registers a cancel for this turn. Returns `false` if the turn is already answering.
+             */
+            fun startCancel(): Boolean {
+                val started = runningCancels.getAndUpdate { if (it == ANSWERING) it else it + 1 } != ANSWERING
+                if (started) _isCancelRequested.value = true
+                return started
+            }
+
+            fun finishCancel() {
+                runningCancels.update { it - 1 }
+            }
+
+            /**
+             * Suspends until no cancel is running for this turn, then marks the turn as answering.
+             */
+            suspend fun awaitCancels() {
+                while (!runningCancels.compareAndSet(0, ANSWERING)) {
+                    runningCancels.first { it == 0 }
+                }
+            }
+
+            private companion object {
+                const val ANSWERING = -1
+            }
+        }
+
         private val _activePrompt = atomic<PromptSession?>(null)
 
+        @OptIn(UnstableApi::class)
         suspend fun prompt(content: List<ContentBlock>, _meta: JsonElement? = null): PromptResponse {
             var response: PromptResponse? = null
             return coroutineScope {
-                try {
-                    val promptJob = launch(start = CoroutineStart.LAZY) {
-                        agentSession.prompt(content, _meta).collect { event ->
-                            when (event) {
-                                is Event.PromptResponseEvent -> {
-                                    if (response != null) {
-                                        logger.error { "Received repeated prompt response: ${event.response} (previous: $response). The last is used" }
-                                    }
-                                    response = event.response
+                val promptJob = launch(start = CoroutineStart.LAZY) {
+                    agentSession.prompt(content, _meta).collect { event ->
+                        when (event) {
+                            is Event.PromptResponseEvent -> {
+                                if (response != null) {
+                                    logger.error { "Received repeated prompt response: ${event.response} (previous: $response). The last is used" }
                                 }
+                                response = event.response
+                            }
 
-                                is Event.SessionUpdateEvent -> {
-                                    clientOperations.notify(event.update, _meta)
-                                }
+                            is Event.SessionUpdateEvent -> {
+                                clientOperations.notify(event.update, _meta)
                             }
                         }
                     }
+                }
 
-                    val promptSession = PromptSession(promptJob)
-                    if (!_activePrompt.compareAndSet(null, promptSession)) {
-                        error("There is already active prompt execution")
-                    }
+                val promptSession = PromptSession(promptJob)
+                if (!_activePrompt.compareAndSet(null, promptSession)) {
+                    error("There is already active prompt execution")
+                }
+                try {
                     promptJob.join()
+                    /*
+                     * A cancelled turn answers only after the session's cancel() returns: the response must follow
+                     * every aborted operation and every update, including those sent from cancel().
+                     */
+                    promptSession.awaitCancels()
+                    /*
+                     * The session's cancel() may stop the turn cooperatively, so its flow ends normally before the
+                     * SDK cancels the job. The client still cancelled the turn and must get the `cancelled` reason.
+                     */
+                    val isCancelled = promptJob.isCancelled || promptSession.isCancelRequested
                     response ?: PromptResponse(
-                        stopReason = if (promptJob.isCancelled) StopReason.CANCELLED else StopReason.END_TURN
+                        stopReason = if (isCancelled) StopReason.CANCELLED else StopReason.END_TURN
                     )
                 } finally {
-                    _activePrompt.getAndSet(null)
+                    // Only the turn frees its slot, so the next prompt can start only after this one has answered.
+                    _activePrompt.value = null
                 }
             }
         }
 
         suspend fun cancel() {
-            // notify AgentSession about upcoming cancellation, this way implementations can gracefully stop ongoing requests
-            agentSession.cancel()
-
-            val activePrompt = _activePrompt.getAndSet(null)
-            if (activePrompt != null) {
-                logger.trace { "Cancelling prompt" }
-                // we expect that all nested outgoing jobs will be cancelled automatically due to structured concurrency
-                // -> prompt task
-                //   <- [request] read file
-                //   -> [response] read file
-                //   <- [request] permissions
-                //   |suspended|
-                // cancelling the whole prompt should cancel all nested outgoing requests. These requests on CE will propagate cancellation to the other side
-                activePrompt.promptJob.cancel()
+            /*
+             * Take the turn to cancel before notifying the session: the session's cancel() may suspend, and a turn
+             * that has already answered may be followed by the next one meanwhile, which this cancel must not touch.
+             */
+            val promptToCancel = _activePrompt.value?.takeIf { it.startCancel() }
+            try {
+                // notify AgentSession about upcoming cancellation, this way implementations can gracefully stop ongoing requests
+                agentSession.cancel()
+
+                if (promptToCancel != null) {
+                    logger.trace { "Cancelling prompt" }
+                    // we expect that all nested outgoing jobs will be cancelled automatically due to structured concurrency
+                    // -> prompt task
+                    //   <- [request] read file
+                    //   -> [response] read file
+                    //   <- [request] permissions
+                    //   |suspended|
+                    // cancelling the whole prompt should cancel all nested outgoing requests. These requests on CE will propagate cancellation to the other side
+                    promptToCancel.promptJob.cancel()
+                }
+            } finally {
+                promptToCancel?.finishCancel()
             }
         }
     }

acp/src/commonMain/kotlin/com/agentclientprotocol/agent/AgentSession.kt

@@ -41,13 +41,28 @@ public interface AgentSession {
      * Sends a message to the agent for execution and waits for the whole turn to be completed.
      * During execution, the agent can send notifications or requests to the client.
      *
+     * The SDK runs at most one turn per session: this method is not called again until the previous turn's
+     * `session/prompt` has been answered, including a cancelled turn that is still winding down. A `session/prompt`
+     * that arrives earlier is rejected with an internal error (`-32603`). The protocol does not define overlapping
+     * prompts; this is an SDK guarantee.
+     *
      * Corresponds to the [AcpMethod.AgentMethods.V1.SessionPrompt]
      */
     public suspend fun prompt(content: List<ContentBlock>, _meta: JsonElement? = null): Flow<Event>
 
     /**
      * Cancels the current agent turn and returns after the agent canceled all activities of the current turn.
      *
+     * The cancelled turn's `session/prompt` is answered only after this method returns, even if the [prompt] flow
+     * ends earlier, so updates sent from here through `currentCoroutineContext().client` reach the client before the
+     * response. This method must therefore not wait for that response. A turn that starts after this call began is
+     * not cancelled by it.
+     *
+     * The cancelled turn is answered with [StopReason.CANCELLED] unless its [prompt] flow emitted a
+     * [Event.PromptResponseEvent], even if this method stops the flow cooperatively and it completes normally.
+     *
+     * Each `session/cancel` calls this method, so a call may start while a previous one is still running.
+     *
      * Corresponds to the [AcpMethod.AgentMethods.V1.SessionCancel]
      */
     public suspend fun cancel() {}