@@ -38,3 +38,23 @@ jobs:
           name: test-results
           path: '**/target/surefire-reports/'
           retention-days: 7
+
+  # Issue #14 only reproduced when the JVM was starved of cores: the agent's lock release
+  # and the client's next prompt race on one CPU. Pin the contention tests to one core so
+  # the window stays open in CI.
+  contention:
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+
+    steps:
+      - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4
+
+      - name: Set up JDK 17
+        uses: actions/setup-java@cf277c60eb25467037889841efdb72551f06f6c3 # v4
+        with:
+          java-version: '17'
+          distribution: 'temurin'
+          cache: maven
+
+      - name: Contention tests pinned to one CPU
+        run: taskset -c 0 ./mvnw -B -pl acp-core -am test -Dtest='MultiPromptSyncSessionTest,AcpAgentSessionPromptLockTest' -Dsurefire.failIfNoSpecifiedTests=false
@@ -22,6 +22,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   connect or start failure instead of dropping it: construction throws `IllegalStateException` when
   the transport refuses synchronously, and every later request fails immediately with the cause.
 
+- **Second prompt on a session could be rejected or hang under CPU contention (#14).** The agent
+  released its single-turn prompt lock in `doFinally`, *after* the response had already reached the
+  client. On a starved machine (`taskset -c 0`, busy CI runners) the client's next prompt arrived
+  before that release and was rejected with `-32000 There is already an active prompt execution`;
+  the rejection's emission then collided with the still-running response emission on the same sink
+  (`FAIL_NON_SERIALIZED`), which the shipped transports silently dropped and the in-memory test
+  transport turned into a fatal error for the agent, so `AcpSyncClient.prompt` waited out the full
+  `requestTimeout`. The lock is now released before the response is published, on the success and
+  error paths alike, and a handler that throws before returning its `Mono` no longer holds the lock
+  forever. Response emission in `StdioAcpAgentTransport`, `WebSocketAcpClientTransport`,
+  `WebSocketAcpAgentTransport` and `InMemoryTransportPair` is serialised the way `sendMessage`
+  already was, and a failed emission in the in-memory pair no longer terminates the agent. The
+  reporter's repro is now a test, and CI runs the contention tests pinned to one CPU. Reported by
+  @krickert.
+- The agent session now reads `sessionId` from typed `PromptRequest`/`CancelNotification` params as
+  well as from maps, so in-process transports get the right lock owner in logs and cancel matching.
+
 ### Added
 
 - `AcpSyncClient(AcpAsyncClient)` is public: the supported way to have both APIs over one session.
 
@@ -158,7 +158,15 @@ private void handleIncomingMessages(Function<Mono<JSONRPCMessage>, Mono<JSONRPCM
 			.flatMap(message -> Mono.just(message).transform(handler))
 			.doOnNext(response -> {
 				if (response != null) {
-					this.outboundSink.tryEmitNext(response);
+					// Responses are emitted from the inbound thread while sendMessage emits
+					// from user threads on the same sink; both must go through the serialising
+					// busy-loop or a collision drops the response (FAIL_NON_SERIALIZED, #14).
+					try {
+						this.outboundSink.emitNext(response, Sinks.EmitFailureHandler.busyLooping(Duration.ofMillis(100)));
+					}
+					catch (Sinks.EmissionException e) {
+						logger.error("Dropped response {}: {}", response, e.getReason());
+					}
 				}
 			})
 			.doOnTerminate(() -> {
 
@@ -167,7 +167,15 @@ private void handleIncomingMessages(Function<Mono<JSONRPCMessage>, Mono<JSONRPCM
 			.flatMap(message -> Mono.just(message).transform(handler))
 			.doOnNext(response -> {
 				if (response != null) {
-					this.outboundSink.tryEmitNext(response);
+					// Responses are emitted from the inbound thread while sendMessage emits
+					// from user threads on the same sink; both must go through the serialising
+					// busy-loop or a collision drops the response (FAIL_NON_SERIALIZED, #14).
+					try {
+						this.outboundSink.emitNext(response, Sinks.EmitFailureHandler.busyLooping(Duration.ofMillis(100)));
+					}
+					catch (Sinks.EmissionException e) {
+						logger.error("Dropped response {}: {}", response, e.getReason());
+					}
 				}
 			})
 			.doOnTerminate(() -> {
 
@@ -275,25 +275,50 @@ private Mono<AcpSchema.JSONRPCResponse> handleIncomingRequest(AcpSchema.JSONRPCR
 							new AcpSchema.JSONRPCError(-32000, "There is already an active prompt execution", null)));
 				}
 
-				// Execute handler and clear active prompt when done
-				return handler.handle(request.params())
+				// The prompt response ends the turn (ACP semantics), so the lock is released
+				// *before* the response is handed downstream to the transport. Releasing in
+				// doFinally instead ran after the response had already reached the client;
+				// under CPU contention the client's next prompt then arrived before the
+				// release and was rejected with -32000 (#14). doOnNext and doOnError run before
+				// the signal propagates, and a Mono emits at most once, so the release is
+				// ordered before publication on every path; doFinally covers cancellation.
+				// Mono.defer keeps a handler that throws synchronously from holding the lock
+				// forever: the throw becomes an error signal that passes through the release.
+				return Mono.defer(() -> handler.handle(request.params()))
 					.map(result -> new AcpSchema.JSONRPCResponse(AcpSchema.JSONRPC_VERSION, request.id(), result, null))
-					.doFinally(signal -> {
-						activePrompt.compareAndSet(newPrompt, null);
-						logger.debug("Prompt completed with signal: {}", signal);
-					});
+					.doOnNext(response -> releasePrompt(newPrompt, "response"))
+					.doOnError(error -> releasePrompt(newPrompt, "error"))
+					.doFinally(signal -> releasePrompt(newPrompt, signal.toString()));
 			}
 
-			return handler.handle(request.params())
+			return Mono.defer(() -> handler.handle(request.params()))
 				.map(result -> new AcpSchema.JSONRPCResponse(AcpSchema.JSONRPC_VERSION, request.id(), result, null));
 		});
 	}
 
 	/**
-	 * Extracts the sessionId from request parameters.
+	 * Releases the single-turn lock if {@code prompt} still holds it. Idempotent: the
+	 * release is attempted on the response, on an error and on the terminal signal, and
+	 * only the first attempt does anything.
+	 */
+	private void releasePrompt(ActivePrompt prompt, String reason) {
+		if (activePrompt.compareAndSet(prompt, null)) {
+			logger.debug("Prompt lock released for sessionId={} requestId={} ({})", prompt.sessionId(),
+					prompt.requestId(), reason);
+		}
+	}
+
+	/**
+	 * 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.
 	 */
-	@SuppressWarnings("unchecked")
 	private String extractSessionId(Object params) {
+		if (params instanceof AcpSchema.PromptRequest promptRequest) {
+			return promptRequest.sessionId() != null ? promptRequest.sessionId() : "unknown";
+		}
+		if (params instanceof AcpSchema.CancelNotification cancelNotification) {
+			return cancelNotification.sessionId() != null ? cancelNotification.sessionId() : "unknown";
+		}
 		if (params instanceof Map<?, ?> map) {
 			Object sessionId = map.get("sessionId");
 			return sessionId != null ? sessionId.toString() : "unknown";
 
@@ -0,0 +1,98 @@
+package com.agentclientprotocol.sdk.integration;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import com.agentclientprotocol.sdk.agent.AcpAgent;
+import com.agentclientprotocol.sdk.agent.AcpSyncAgent;
+import com.agentclientprotocol.sdk.client.AcpClient;
+import com.agentclientprotocol.sdk.client.AcpSyncClient;
+import com.agentclientprotocol.sdk.spec.AcpSchema;
+import com.agentclientprotocol.sdk.test.InMemoryTransportPair;
+import org.junit.jupiter.api.RepeatedTest;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Reproduction for issue #14: AcpSyncAgent + AcpSyncClient over InMemoryTransportPair, one
+ * session, three sequential prompts, run under {@code taskset -c 0}. The report says the
+ * second or third prompt never returns and the request timeout fires.
+ */
+class MultiPromptSyncSessionTest {
+
+	private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+	@RepeatedTest(5)
+	void threeSequentialPromptsOnOneSessionAllReturn() {
+		InMemoryTransportPair pair = InMemoryTransportPair.create();
+		AcpSyncAgent agent = AcpAgent.sync(pair.agentTransport())
+			.requestTimeout(TIMEOUT)
+			.initializeHandler(request -> new AcpSchema.InitializeResponse(1, new AcpSchema.AgentCapabilities(), List.of()))
+			.newSessionHandler(request -> new AcpSchema.NewSessionResponse("session-14", null, null))
+			.promptHandler((request, context) -> new AcpSchema.PromptResponse(AcpSchema.StopReason.END_TURN))
+			.build();
+		agent.start();
+
+		AcpSyncClient client = AcpClient.sync(pair.clientTransport()).requestTimeout(TIMEOUT).build();
+		try {
+			client.initialize();
+			String sessionId = client.newSession(new AcpSchema.NewSessionRequest("/workspace", List.of())).sessionId();
+			for (int i = 1; i <= 3; i++) {
+				long t0 = System.nanoTime();
+				AcpSchema.PromptResponse response = client.prompt(new AcpSchema.PromptRequest(sessionId,
+						List.of(new AcpSchema.TextContent("prompt " + i))));
+				System.out.printf("prompt %d returned %s in %d ms%n", i, response.stopReason(),
+						(System.nanoTime() - t0) / 1_000_000);
+				assertThat(response.stopReason()).isEqualTo(AcpSchema.StopReason.END_TURN);
+			}
+		}
+		finally {
+			client.closeGracefully();
+			agent.closeGracefully();
+			pair.closeGracefully().block(TIMEOUT);
+		}
+	}
+
+	@RepeatedTest(5)
+	void threeSequentialStreamingPromptsOnOneSessionAllReturn() {
+		InMemoryTransportPair pair = InMemoryTransportPair.create();
+		AcpSyncAgent agent = AcpAgent.sync(pair.agentTransport())
+			.requestTimeout(TIMEOUT)
+			.initializeHandler(request -> new AcpSchema.InitializeResponse(1, new AcpSchema.AgentCapabilities(), List.of()))
+			.newSessionHandler(request -> new AcpSchema.NewSessionResponse("session-14", null, null))
+			.promptHandler((request, context) -> {
+				for (int chunk = 0; chunk < 3; chunk++) {
+					context.sendUpdate(request.sessionId(),
+							new AcpSchema.AgentMessageChunk("agent_message_chunk", new AcpSchema.TextContent("chunk " + chunk)));
+				}
+				return new AcpSchema.PromptResponse(AcpSchema.StopReason.END_TURN);
+			})
+			.build();
+		agent.start();
+
+		List<String> updates = new CopyOnWriteArrayList<>();
+		AcpSyncClient client = AcpClient.sync(pair.clientTransport())
+			.requestTimeout(TIMEOUT)
+			.sessionUpdateConsumer(notification -> updates.add(notification.update().getClass().getSimpleName()))
+			.build();
+		try {
+			client.initialize();
+			String sessionId = client.newSession(new AcpSchema.NewSessionRequest("/workspace", List.of())).sessionId();
+			for (int i = 1; i <= 3; i++) {
+				long t0 = System.nanoTime();
+				AcpSchema.PromptResponse response = client.prompt(new AcpSchema.PromptRequest(sessionId,
+						List.of(new AcpSchema.TextContent("prompt " + i))));
+				System.out.printf("streaming prompt %d returned %s in %d ms (%d updates so far)%n", i,
+						response.stopReason(), (System.nanoTime() - t0) / 1_000_000, updates.size());
+				assertThat(response.stopReason()).isEqualTo(AcpSchema.StopReason.END_TURN);
+			}
+		}
+		finally {
+			client.closeGracefully();
+			agent.closeGracefully();
+			pair.closeGracefully().block(TIMEOUT);
+		}
+	}
+
+}