Model Reactor's one-shot cancel in the PendingResponses Lincheck sink

The lincheck job (scale 10) failed with "request 2 caused 0
$/cancel_request, cancelled before any signal: true". The interleaving:
two threads cancel request 2. The first runs the cancel callback,
remembers the id and is preempted before its pending.remove. The second
finds the cancel callback already taken, runs the dispose callback, and
that removes the entry; the first's remove then fails, so no cancel is
reported.

Reactor cannot produce this. MonoCreate's sink swaps its state once in
cancel() and runs onCancel then onDispose on the winning thread; a second
cancel is a no-op. The model's RecordingSink let a second cancel run the
dispose callback. It now acts on the first cancel only, and a plain test
pins the Reactor contract it relies on.

Green at scale 10 and 20. The model still fails each planted bug: a
cancel also sent from the dispose path, a cancel sent without the removal
check, and an entry left behind.

Author: Mark Pollack

acp-core/src/test/java/com/agentclientprotocol/sdk/spec/PendingResponsesLincheckTest.java

@@ -5,7 +5,14 @@
 package com.agentclientprotocol.sdk.spec;
 
 import java.util.HashSet;
+import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicIntegerArray;
 import java.util.concurrent.atomic.AtomicReference;
@@ -23,10 +30,15 @@
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.reactivestreams.Subscription;
 import reactor.core.Disposable;
+import reactor.core.publisher.BaseSubscriber;
+import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoSink;
 import reactor.util.context.Context;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 /**
  * Model checks {@link PendingResponses}, the table of requests waiting for a response, with
  * Lincheck. Requests are registered, answered, cancelled (their sink is disposed, as a
@@ -84,6 +96,62 @@ void noRequestIsLostOrCompletedTwice() {
 			.check(Requests.class);
 	}
 
+	/**
+	 * The Reactor contract the model's sink relies on: a cancel that arrives while another is
+	 * still in the cancel callback does nothing, so the dispose callback (which removes the
+	 * entry) cannot overtake the cancel callback (which reports the cancel only if it removes
+	 * the entry itself).
+	 */
+	@Test
+	void reactorRunsTheCallbacksOfOnlyTheFirstCancel() throws Exception {
+		CountDownLatch inCancelCallback = new CountDownLatch(1);
+		CountDownLatch release = new CountDownLatch(1);
+		AtomicInteger cancelCallbacks = new AtomicInteger();
+		AtomicBoolean disposed = new AtomicBoolean();
+		AtomicReference<@Nullable Subscription> subscription = new AtomicReference<>();
+		Mono.create(sink -> {
+			sink.onCancel(() -> {
+				cancelCallbacks.incrementAndGet();
+				inCancelCallback.countDown();
+				awaitQuietly(release);
+			});
+			sink.onDispose(() -> disposed.set(true));
+		}).subscribe(new BaseSubscriber<Object>() {
+			@Override
+			protected void hookOnSubscribe(Subscription s) {
+				subscription.set(s);
+			}
+		});
+		Subscription sink = Objects.requireNonNull(subscription.get());
+		ExecutorService first = Executors.newSingleThreadExecutor();
+		try {
+			Future<?> firstCancel = first.submit(sink::cancel);
+			assertThat(inCancelCallback.await(5, TimeUnit.SECONDS)).isTrue();
+
+			sink.cancel();
+
+			assertThat(disposed).as("the second cancel ran the dispose callback").isFalse();
+			release.countDown();
+			firstCancel.get(5, TimeUnit.SECONDS);
+			assertThat(cancelCallbacks).hasValue(1);
+			assertThat(disposed).isTrue();
+		}
+		finally {
+			release.countDown();
+			first.shutdownNow();
+		}
+	}
+
+	private static void awaitQuietly(CountDownLatch latch) {
+		try {
+			if (!latch.await(5, TimeUnit.SECONDS)) {
+				throw new IllegalStateException("not released");
+			}
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+		}
+	}
 
 	@Param(name = "id", gen = IntGen.class, conf = "1:2")
 	public static class Requests {
@@ -218,13 +286,28 @@ static final class RecordingSink implements MonoSink<AcpSchema.JSONRPCResponse>
 
 		volatile boolean cancelled;
 
+		/** Whether a cancel already ran: Reactor's sink acts on the first cancel only. */
+		private final AtomicBoolean cancelRan = new AtomicBoolean();
+
 		private final AtomicReference<@Nullable Disposable> onDispose = new AtomicReference<>();
 
 		private final AtomicReference<@Nullable Disposable> onCancel = new AtomicReference<>();
 
-		/** Like Reactor's: the cancel callback, then the dispose callback, unless already signalled. */
+		/**
+		 * Like Reactor's: the cancel callback, then the dispose callback, unless already
+		 * signalled. Only the first cancel does anything, as in Reactor's sink, whose
+		 * {@code cancel()} swaps its state once and runs both callbacks on the thread that
+		 * won: a second cancel (the caller disposes as its timeout fires, say) cannot run the
+		 * dispose callback while the first is still in the cancel callback. A model that let
+		 * it would report a lost {@code $/cancel_request} that Reactor never produces
+		 * ({@link PendingResponsesLincheckTest#reactorRunsTheCallbacksOfOnlyTheFirstCancel()} pins
+		 * that contract).
+		 */
 		void cancel() {
 			cancelled = true;
+			if (!cancelRan.compareAndSet(false, true)) {
+				return;
+			}
 			Disposable callback = onCancel.getAndSet(null);
 			if (callback != null && signals.get() == 0) {
 				callback.dispose();