Cancel requests both ways with $/cancel_request

ACP v1 Cancellation (stable since 2026-06-29), in the client and agent
sessions alike.

Sending: disposing a request's Mono before its response (the caller,
timeout(), or the SDK request timeout) sends $/cancel_request once, after
the request itself was written. Never for a request already answered,
dismissed by close, or that failed to send; a subscriber that cancels
before the request is created sends nothing. The report comes from
MonoSink.onCancel, not onDispose, which also runs after an error. The
peer's late answer is expected and logged at DEBUG (the last 1,024
cancelled ids are remembered). The send runs on the timeout delivery
threads, never on the shared timer. Each subscription to a request now
gets its own id: a retry used to reuse it, and the first response
completed both.

Receiving: InboundRequests wraps a request's complete answer, so a
cancel and the handler's answer race inside one Mono (takeUntilOther):
exactly one response, the handler's or -32800. The request is cancellable
only once its handler is subscribed, and no longer once its value is
produced. Once cancelled, a handler error (an interrupted blocking
handler throws) counts as cancelled. A cancelled prompt ends its turn
before the answer is published; one already cancelled by session/cancel
answers stop reason cancelled instead of -32800, as does a prompt handler
that fails with a cancellation after session/cancel. $/cancel_request is
handled by the session, never offered to notification handlers, and on
the client not queued behind the ordered notification drain. Unknown and
answered ids are ignored; integer ids match whatever integer type the
mapper produced, and "7" is not 7.

Graceful cancel: RequestCancellation.cancelWhen(trigger), written into
a request's Reactor context, sends $/cancel_request when the trigger
fires but keeps waiting, so the request ends with the peer's answer
(its result or AcpError -32800). At most one $/cancel_request per
request, whichever way it is cancelled.

Internal cancellation: CancellationException, an interrupt, or an
AcpProtocolException with -32800 from a handler is answered -32800.
Closing a session cancels the requests it is still handling.

Streamable HTTP: $/cancel_request is connection-scoped both ways, like
every $/ method (the routing tables' rule, and the Rust SDK's): the
client posts it without Acp-Session-Id, the agent sends it on the
connection stream even when its params name a session. A cancel that
overtakes its request on another stream finds no request and is
ignored, as the spec allows. A received $/cancel_request whose
requestId is not a string or integer is ignored: a notification has no
response to carry -32602. CancelRequestNotification carries _meta, and
its MetaCoverageTest exclusion is gone.

Also corrects the G0f note that a blocked sync handler is not
interrupted: the sync handler schedulers interrupt on cancel
(CHANGELOG, AcpAgent, AcpAgentSupport and @cancel Javadoc).

Cross-SDK suite: the Java interop client and agent run the three
cancel-request steps (client, agent, unknown id) and their P2 expected
failures are deleted; they pass Java<->Java and Java<->TypeScript on
stdio, Streamable HTTP and WebSocket, and Java<->Python where Python
can. Two Python peer gaps are declared: the Python client routes no
$/cancel_request, so an agent's cancelled fs read is never answered
-32800 and the client's own step sees the prompt end late.

Tests: CancelRequestTest (session level, both sides), an end-to-end test
in both directions plus a graceful cancel (red on main), InboundRequestsLincheckTest (exactly
one response, no table leak), and PendingResponsesLincheckTest extended
(a cancel is sent iff the caller gave up before any signal, at most
once). Each model fails on a planted bug.

Author: Mark Pollack

CHANGELOG.md

@@ -15,6 +15,43 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **`$/cancel_request` (ACP v1, Cancellation), both directions.** Client and agent sessions alike:
+  - **Cancelling a request you sent:** dispose (cancel) the subscription to its `Mono` before the
+    response arrives, directly or through `timeout(...)`, `take...`, or the SDK's own request
+    timeout, and the session sends `$/cancel_request {"requestId": ...}` to the peer, once, after
+    the request itself was written (never for a request that failed to send, was already answered,
+    or was dismissed by closing the session). The caller's `Mono` is cancelled at once (Reactor
+    semantics); the peer's late answer is expected and discarded at DEBUG. An agent prompt handler
+    whose subscription is cancelled cancels the `fs/*`, `terminal/*` and
+    `session/request_permission` requests it is waiting on the same way. Interrupting a thread
+    blocked in a sync client or agent call cancels its request too.
+  - **Graceful cancel:** `request.contextWrite(RequestCancellation.cancelWhen(trigger))` sends
+    `$/cancel_request` when `trigger` emits or completes but keeps waiting, so the request ends with
+    the peer's answer: its result, or an `AcpError` with code `-32800`. Works for every typed request
+    method on both sides; at most one `$/cancel_request` per request either way.
+  - **Answering a cancelled request:** on `$/cancel_request` the session cancels the handler's
+    subscription and answers the request with error `-32800` (`Request cancelled`), unless the
+    handler answered first; exactly one response either way (model checked with Lincheck,
+    `InboundRequestsLincheckTest`). An unknown or already answered id is ignored. A cancelled
+    `session/prompt` ends its turn before the answer is published; one already cancelled with
+    `session/cancel` is answered with stop reason `cancelled` instead, as ACP v1 requires of a
+    cancelled prompt. `$/cancel_request` is handled by the session and never reaches notification
+    handlers; on the client it is not queued behind slow `session/update` consumers.
+  - **Internal cancellation:** a handler that fails with `CancellationException`, an interrupt, or
+    an `AcpProtocolException` with code `-32800` is answered `-32800` (a prompt being cancelled:
+    stop reason `cancelled`). Closing a session cancels the requests it is still handling and
+    answers them `-32800` where the transport still delivers.
+  - Over Streamable HTTP, `$/cancel_request` is connection-scoped both ways, like every `$/`
+    method and as in the Rust SDK: the client posts it without `Acp-Session-Id`, the agent sends
+    it on the connection stream. It can therefore overtake a request still on a session stream;
+    the receiver then ignores it as an unknown id (the spec's MAY) and the request completes.
+  - Params of a received `$/cancel_request` without a string or integer `requestId` are ignored
+    (a notification has no response to carry `-32602`).
+  - `AcpSchema.METHOD_CANCEL_REQUEST`, `AcpSchema.CancelRequestNotification`,
+    `RequestCancellation`, `AcpSchedulers.timeoutDelivery()`.
+  - Cross-SDK suite: the Java programs run `cancel-request.client`, `cancel-request.agent` and
+    `cancel-request.unknown`; their `P2` expected failures are gone.
+
 - **Prompt timeouts on the agent builders** (`AcpAgent.async(...)`, `AcpAgent.sync(...)` and
   `AcpAgentSupport.builder()`), also as `PromptTimeouts` on the `AcpAgentSession` constructor:
   - `cancelGracePeriod(Duration)`: how long a prompt handler has to answer after `session/cancel`
@@ -30,9 +67,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   answer leaves through the normal path: the turn ends just before it is published, it follows
   every update the handler had already sent, and exactly one answer is sent when the handler
   answers just as a timer fires (model checked with Lincheck, `PromptAnswerLincheckTest`). The
-  timers run on the SDK's shared timeout timer, with no new threads. A handler that keeps running
-  after its subscription is cancelled (a blocking sync handler is not interrupted) and sends more
-  updates sends them after the answer. `AcpErrorCodes.REQUEST_CANCELLED` (`-32800`) is new.
+  timers run on the SDK's shared timeout timer, with no new threads. Cancelling a sync handler
+  interrupts its thread (the SDK's sync handler scheduler interrupts on cancel); a handler that
+  ignores the interrupt, or any handler that keeps running after its subscription is cancelled,
+  and sends more updates sends them after the answer. `AcpErrorCodes.REQUEST_CANCELLED` (`-32800`) is new.
 - **`session_info_update`** (`AcpSchema.SessionInfoUpdate`): the agent tells the client the
   session's title and last activity time. Stable in ACP v1. Known limit: the schema lets a peer
   send `null` to clear a field; the record reads an explicit `null` like a missing field and never
@@ -71,7 +109,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   common permission-request case (migration: use it, or add `null` for `_meta`). A test walks a copy of the v1.9.1 schema
   (`acp-core/src/test/resources/schema/v1/schema.json`) and fails on any object with `_meta`
   whose record lacks it. Not covered yet: the elicitation records (with the elicitation work), the
-  presence markers typed `Object` (they keep `_meta` as a map entry) and `$/cancel_request`.
+  presence markers typed `Object` (they keep `_meta` as a map entry). `CancelRequestNotification`
+  carries `_meta` (with `$/cancel_request`, below).
 
 ### Changed
 
@@ -156,6 +195,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   constructor. Migration: `new AgentCapabilities(loadSession, session, mcp, prompt, providers,
   meta)` becomes `new AgentCapabilities(loadSession, session, mcp, prompt, auth, providers, meta)`
   (`auth` may be `null`); the shorter constructors are unchanged.
+- **Behaviour change: a request whose caller gives up is cancelled at the peer.** Disposing a
+  request `Mono`, or its timeout firing (the client's default request timeout is 30 seconds, the
+  agent's 60), now sends `$/cancel_request`, so a peer that supports it stops the work and answers
+  `-32800`. A `session/prompt` that times out on the client is therefore cancelled at an agent that
+  supports `$/cancel_request` (the Java agent answers it at once and ends the turn), where before
+  the agent kept running the turn. Migration: raise `requestTimeout` for long prompts, or send
+  `session/cancel` for the graceful prompt cancel with stop reason `cancelled`.
+- **Fix: a resubscribed request is a new request.** Subscribing twice to one `sendRequest` `Mono`
+  (a `retry()`, say) reused the request id, so the first response completed both. Each
+  subscription now gets its own id.
 - **Behaviour change: `session/cancel` no longer ends the prompt turn; the cancelled prompt's
   response does.** The agent session used to free the session for a new prompt as soon as the
   cancel notification arrived, so a client could start a second prompt while the cancelled one's

acp-agent-support/src/main/java/com/agentclientprotocol/sdk/agent/support/AcpAgentSupport.java

@@ -463,8 +463,8 @@ public Builder requestTimeout(Duration timeout) {
 
 		/**
 		 * Set how long a {@code @Prompt} method has to return after {@code session/cancel}
-		 * before the agent answers the prompt with stop reason {@code cancelled} itself. The
-		 * method's thread is not interrupted. Default 60 seconds; {@link Duration#ZERO} for
+		 * before the agent answers the prompt with stop reason {@code cancelled} itself and
+		 * interrupts the method's thread. Default 60 seconds; {@link Duration#ZERO} for
 		 * none. See {@code AcpAgent.SyncAgentBuilder#cancelGracePeriod}.
 		 * @param gracePeriod the grace period; zero for none, not negative
 		 * @return this builder

acp-annotations/src/main/java/com/agentclientprotocol/sdk/annotation/Cancel.java

@@ -23,8 +23,8 @@
  * send any last updates, and return stop reason {@code cancelled}. Until it has returned
  * (or failed), the session rejects a new prompt (ACP v1, prompt turn, Cancellation). If it
  * has not returned within the cancel grace period (60 seconds unless configured on the
- * agent builder), the agent answers the prompt {@code cancelled} itself; the method's
- * thread is not interrupted.
+ * agent builder), the agent answers the prompt {@code cancelled} itself and interrupts the
+ * method's thread.
  *
  * <p>The method can have the following parameters (all optional):
  * <ul>

acp-core/src/main/java/com/agentclientprotocol/sdk/agent/AcpAgent.java

@@ -839,7 +839,7 @@ public SyncAgentBuilder requestTimeout(Duration timeout) {
 
 		/**
 		 * Sets how long a prompt handler has to answer after {@code session/cancel}. When it
-		 * passes, the agent cancels the handler (a sync handler blocked on its thread is not interrupted; what it sends afterwards is its own) and answers the prompt itself with stop
+		 * passes, the agent cancels the handler (a sync handler blocked on its thread is interrupted; what it sends if it keeps running is its own) and answers the prompt itself with stop
 		 * reason {@code cancelled}, as ACP requires of a cancelled prompt, which ends the
 		 * turn so the session accepts a new prompt. Updates the handler sent before that
 		 * answer reach the client first. Default: 60 seconds

acp-core/src/main/java/com/agentclientprotocol/sdk/client/transport/StreamableHttpRoutes.java

@@ -78,7 +78,7 @@ private enum MethodScope {
 			Map.entry(AcpSchema.METHOD_PROVIDERS_SET, MethodScope.CONNECTION),
 			Map.entry(AcpSchema.METHOD_PROVIDERS_DISABLE, MethodScope.CONNECTION),
 			Map.entry("nes/start", MethodScope.CONNECTION), Map.entry("mcp/message", MethodScope.CONNECTION),
-			Map.entry("$/cancel_request", MethodScope.CONNECTION),
+			Map.entry(AcpSchema.METHOD_CANCEL_REQUEST, MethodScope.CONNECTION),
 			Map.entry(AcpSchema.METHOD_SESSION_LOAD, MethodScope.SESSION_ANSWERED_ON_CONNECTION),
 			Map.entry(AcpSchema.METHOD_SESSION_RESUME, MethodScope.SESSION_ANSWERED_ON_CONNECTION),
 			Map.entry(AcpSchema.METHOD_SESSION_PROMPT, MethodScope.SESSION),

acp-core/src/main/java/com/agentclientprotocol/sdk/spec/AcpAgentSession.java

@@ -9,6 +9,7 @@
 import java.util.Set;
 import java.util.function.Function;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
 
 
 import com.agentclientprotocol.sdk.error.AcpErrorCodes;
@@ -60,6 +61,9 @@ public class AcpAgentSession implements AcpSession {
 	/** Map of notification handlers keyed by method name */
 	private final ConcurrentHashMap<String, NotificationHandler> notificationHandlers = new ConcurrentHashMap<>();
 
+	/** Requests from the client still being handled, which a $/cancel_request can cancel */
+	private final InboundRequests inbound = new InboundRequests();
+
 	/** Single-turn enforcement: the active prompt of each logical ACP sessionId. */
 	private final ActivePrompts activePrompts = new ActivePrompts();
 
@@ -197,7 +201,11 @@ private Mono<AcpSchema.JSONRPCMessage> handle(AcpSchema.JSONRPCMessage message)
 		if (message instanceof AcpSchema.JSONRPCRequest request) {
 			logger.debug("Received request method={} id={}", request.method(), request.id());
 			// Mono.from widens the response Mono to the message type without an operator.
-			return Mono.from(handleIncomingRequest(request)
+			// The whole answer is tracked, so a $/cancel_request races it inside one Mono.
+			AtomicReference<ActivePrompts.@Nullable Turn> promptTurn = new AtomicReference<>();
+			return Mono.from(this.inbound
+				.track(request, handleIncomingRequest(request, promptTurn),
+						() -> whenCancelled(request, promptTurn.get()))
 				.onErrorResume(error -> Mono.just(InboundMessages.error(request, error))));
 		}
 		if (message instanceof AcpSchema.JSONRPCNotification notification) {
@@ -214,7 +222,8 @@ private Mono<AcpSchema.JSONRPCMessage> handle(AcpSchema.JSONRPCMessage message)
 	 * @param request The incoming JSON-RPC request
 	 * @return A Mono containing the JSON-RPC response
 	 */
-	private Mono<AcpSchema.JSONRPCResponse> handleIncomingRequest(AcpSchema.JSONRPCRequest request) {
+	private Mono<AcpSchema.JSONRPCResponse> handleIncomingRequest(AcpSchema.JSONRPCRequest request,
+			AtomicReference<ActivePrompts.@Nullable Turn> promptTurn) {
 		return Mono.defer(() -> {
 			var handler = this.requestHandlers.get(request.method());
 			if (handler == null) {
@@ -233,17 +242,22 @@ private Mono<AcpSchema.JSONRPCResponse> handleIncomingRequest(AcpSchema.JSONRPCR
 							"There is already an active prompt on session " + sessionId,
 							Map.of("sessionId", sessionId)));
 				}
+				promptTurn.set(turn);
 
 				// The turn ends before the response is published (#14): see
 				// ActivePrompts.endBeforePublishing. Mono.defer keeps a handler that throws
 				// synchronously from holding the turn forever: the throw becomes an error
 				// signal that passes through the release.
 				// The deadlines may answer instead of the handler; either answer passes
 				// through the release.
+				// A handler whose work was aborted after session/cancel answers cancelled, not
+				// an error (ACP v1, prompt turn, Cancellation).
 				return activePrompts.endBeforePublishing(turn, promptDeadlines.answer(turn, request,
 						InboundMessages.requireResult(Mono.defer(() -> handler.handle(InboundMessages.paramsOrEmpty(request.params()))),
 								request.method())
-							.map(result -> InboundMessages.result(request, result))));
+							.map(result -> InboundMessages.result(request, result))
+							.onErrorResume(error -> InboundMessages.isCancellation(error) && turn.answer().isCancelling(),
+									error -> Mono.just(cancelledPrompt(request)))));
 			}
 
 			return InboundMessages.requireResult(Mono.defer(() -> handler.handle(InboundMessages.paramsOrEmpty(request.params()))),
@@ -252,6 +266,23 @@ private Mono<AcpSchema.JSONRPCResponse> handleIncomingRequest(AcpSchema.JSONRPCR
 		});
 	}
 
+	/**
+	 * The answer to a request a $/cancel_request cancelled: -32800, except for a prompt
+	 * already cancelled by session/cancel, which must answer stop reason cancelled (ACP v1,
+	 * prompt turn, Cancellation).
+	 */
+	private static AcpSchema.JSONRPCResponse whenCancelled(AcpSchema.JSONRPCRequest request,
+			ActivePrompts.@Nullable Turn promptTurn) {
+		if (promptTurn != null && promptTurn.answer().isCancelling()) {
+			return cancelledPrompt(request);
+		}
+		return InboundRequests.cancelled(request);
+	}
+
+	private static AcpSchema.JSONRPCResponse cancelledPrompt(AcpSchema.JSONRPCRequest request) {
+		return InboundMessages.result(request, new AcpSchema.PromptResponse(AcpSchema.StopReason.CANCELLED));
+	}
+
 	/**
 	 * Extracts the sessionId from request parameters, which arrive as a {@code Map} from
 	 * the JSON transports and as the typed record from in-process transports.
@@ -282,12 +313,20 @@ private MethodNotFoundError getMethodNotFoundError(String method) {
 	/**
 	 * Handles an incoming JSON-RPC notification by routing it to the appropriate handler.
 	 * A session/cancel notification does not end the active prompt's turn: the cancelled
-	 * prompt's response does (see ActivePrompts#cancel).
+	 * prompt's response does (see ActivePrompts#cancel). A $/cancel_request cancels the
+	 * request it names, which is then answered with -32800 unless it already answered; a
+	 * cancelled prompt's turn ends before that answer is published.
 	 * @param notification The incoming JSON-RPC notification
 	 * @return A Mono that completes when the notification is processed
 	 */
 	private Mono<Void> handleIncomingNotification(AcpSchema.JSONRPCNotification notification) {
 		return Mono.defer(() -> {
+			// $/cancel_request: answered by the session itself, never by a handler
+			if (AcpSchema.METHOD_CANCEL_REQUEST.equals(notification.method())) {
+				this.inbound.cancel(notification);
+				return Mono.empty();
+			}
+
 			// session/cancel: the turn stays until the cancelled prompt answers
 			if (AcpSchema.METHOD_SESSION_CANCEL.equals(notification.method())) {
 				activePrompts.cancel(extractSessionId(notification.params()));
@@ -371,6 +410,7 @@ public Set<String> getActivePromptSessionIds() {
 	@Override
 	public Mono<Void> closeGracefully() {
 		return Mono.fromRunnable(() -> {
+			this.inbound.cancelAll();
 			activePrompts.clear();
 			dismissPendingResponses();
 		}).then(this.transport.closeGracefully());
@@ -381,6 +421,7 @@ public Mono<Void> closeGracefully() {
 	 */
 	@Override
 	public void close() {
+		this.inbound.cancelAll();
 		activePrompts.clear();
 		dismissPendingResponses();
 		transport.close();

acp-core/src/main/java/com/agentclientprotocol/sdk/spec/AcpClientSession.java

@@ -64,6 +64,9 @@ public class AcpClientSession implements AcpSession {
 	/** Requests and notifications sent to the agent, and the requests waiting for a response */
 	private final OutboundMessages outbound;
 
+	/** Requests from the agent still being handled, which a $/cancel_request can cancel */
+	private final InboundRequests inbound = new InboundRequests();
+
 	/** Map of request handlers keyed by method name */
 	private final ConcurrentHashMap<String, RequestHandler<?>> requestHandlers = new ConcurrentHashMap<>();
 
@@ -233,18 +236,29 @@ else if (message instanceof AcpSchema.JSONRPCRequest request) {
 			respondTo(request);
 		}
 		else if (message instanceof AcpSchema.JSONRPCNotification notification) {
-			enqueue(notification);
+			if (AcpSchema.METHOD_CANCEL_REQUEST.equals(notification.method())) {
+				// Not queued behind the ordered notification drain: a slow session/update
+				// consumer must not delay cancelling a pending request.
+				this.inbound.cancel(notification);
+			}
+			else {
+				enqueue(notification);
+			}
 		}
 		else {
 			logger.warn("Received unknown message type: {}", message.getClass().getName());
 		}
 	}
 
-	/** Answers a request from the agent; a failed handler is answered with an error response. */
+	/**
+	 * Answers a request from the agent; a failed handler is answered with an error response,
+	 * a request cancelled by $/cancel_request with -32800 unless it already answered.
+	 */
 	private void respondTo(AcpSchema.JSONRPCRequest request) {
 		logger.debug("Received request method={} id={}", request.method(), request.id());
 		logger.trace("Incoming request method='{}' id={}", request.method(), request.id());
-		handleIncomingRequest(request).onErrorResume(error -> Mono.just(InboundMessages.error(request, error)))
+		this.inbound.track(request, handleIncomingRequest(request))
+			.onErrorResume(error -> Mono.just(InboundMessages.error(request, error)))
 			.flatMap(this.transport::sendMessage)
 			.onErrorComplete(t -> {
 				logger.warn("Issue sending response to the agent, ", t);
@@ -376,6 +390,7 @@ public Mono<Void> sendNotification(String method, @Nullable Object params) {
 	public Mono<Void> closeGracefully() {
 		return Mono.<Void>fromRunnable(() -> {
 			this.closing = true;
+			this.inbound.cancelAll();
 			dismissPendingResponses();
 			// Never lost to a notification being delivered concurrently (NotificationQueue).
 			this.notifications.complete();
@@ -396,6 +411,7 @@ public Mono<Void> closeGracefully() {
 	@Override
 	public void close() {
 		this.closing = true;
+		this.inbound.cancelAll();
 		dismissPendingResponses();
 		this.notifications.complete();
 		notificationSubscription.dispose();