Drop prompt context updates once the prompt has been answered

ACP requires a prompt's session/update notifications to precede its
answer, also after session/cancel. A handler still running after the SDK
answered the prompt (cancel grace period or maxPromptDuration), or one
sending after its own answer, sent updates after the answer. The prompt
contexts now ask the prompt's turn, through the answered-once state that
decides who answers, and drop such updates with a DEBUG log.

Author: Mark Pollack

CHANGELOG.md

@@ -241,7 +241,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   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.
+  has the updates it then sends through its prompt context dropped (see Fixed). `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
@@ -1043,6 +1043,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **A prompt's updates no longer follow its answer.** ACP requires an agent to send a prompt's
+  `session/update` notifications before it answers the prompt, also after `session/cancel`. When
+  the SDK answered a prompt itself (the cancel grace period or `maxPromptDuration` passed), a
+  handler that kept running could still send updates through its prompt context, and they went out
+  after the answer; so could a handler that sent through its context after returning its own
+  answer. Once a prompt has been answered, by its handler or by the SDK, `PromptContext.sendUpdate`
+  and `SyncPromptContext.sendUpdate` (and the helpers built on them, such as `sendMessage`) drop
+  the update and log it at DEBUG, without its content; the `Mono` completes empty. The check uses
+  the prompt's answered-once state that decides who answers. Updates sent with
+  `sendSessionUpdate` outside a prompt context are not affected.
+
 - **A builder agent advertises `providers` for any provider handler.** Without an initialize
   handler, a builder agent advertised `providers` only for a `providers/list` handler, while an
   annotated agent does for any of the three provider methods; with only a set or disable handler

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

@@ -1120,8 +1120,9 @@ 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, which interrupts its thread if it is blocked (what
-		 * it sends if it keeps running is its own), and answers the prompt itself with stop reason
+		 * passes, the agent cancels the handler, which interrupts its thread if it is blocked (updates
+		 * it sends through its prompt context if it keeps running are dropped), and answers the
+		 * prompt itself with stop reason
 		 * {@code cancelled}, as ACP requires of a cancelled prompt; that ends the turn, so the
 		 * session accepts a new prompt. Updates the handler sent before that answer reach the
 		 * client first. Default: 60 seconds ({@link PromptTimeouts#DEFAULT_CANCEL_GRACE_PERIOD});

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

@@ -9,6 +9,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BooleanSupplier;
 import java.util.function.Predicate;
 
 import com.agentclientprotocol.sdk.capabilities.NegotiatedCapabilities;
@@ -149,6 +150,15 @@ PromptCancellations.Signal promptSignal(String sessionId) {
 		return (signal != null) ? signal : new PromptCancellations.Signal();
 	}
 
+	/**
+	 * Whether the prompt running on {@code sessionId} has been answered; always false when no
+	 * prompt is running (a context built outside a running prompt).
+	 */
+	BooleanSupplier promptAnswered(String sessionId) {
+		AcpAgentSession current = this.session;
+		return (current != null) ? current.promptAnswered(sessionId) : () -> false;
+	}
+
 	private <T> AcpAgentSession.NotificationHandler sessionHandler(AgentHandlers.Notification<T> registration) {
 		return params -> registration.handler()
 			.apply(transport.unmarshalParams(params, registration.notificationType()));

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

@@ -9,6 +9,7 @@
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BooleanSupplier;
 
 import com.agentclientprotocol.sdk.capabilities.NegotiatedCapabilities;
 import com.agentclientprotocol.sdk.spec.AcpError;
@@ -58,6 +59,9 @@ class DefaultPromptContext implements PromptContext {
 
 	private final PromptCancellations.Signal cancellation;
 
+	/** Whether this context's prompt has been answered: its updates are then dropped. */
+	private final BooleanSupplier answered;
+
 	/**
 	 * Creates a new prompt context wrapping the given agent.
 	 * @param agent The agent to delegate to
@@ -68,6 +72,8 @@ class DefaultPromptContext implements PromptContext {
 		this.sessionId = sessionId;
 		this.cancellation = (agent instanceof DefaultAcpAsyncAgent running) ? running.promptSignal(sessionId)
 				: new PromptCancellations.Signal();
+		this.answered = (agent instanceof DefaultAcpAsyncAgent running) ? running.promptAnswered(sessionId)
+				: () -> false;
 	}
 
 	// ========================================================================
@@ -76,7 +82,15 @@ class DefaultPromptContext implements PromptContext {
 
 	@Override
 	public Mono<Void> sendUpdate(AcpSchema.SessionUpdate update) {
-		return agent.sendSessionUpdate(sessionId, update);
+		return Mono.defer(() -> {
+			if (this.answered.getAsBoolean()) {
+				// ACP: a prompt's session/update notifications must precede its answer.
+				logger.debug("Dropped a {} update for session {}: its prompt has been answered",
+						update.getClass().getSimpleName(), sessionId);
+				return Mono.empty();
+			}
+			return agent.sendSessionUpdate(sessionId, update);
+		});
 	}
 
 	@Override
@@ -213,14 +227,13 @@ private Mono<AcpSchema.RequestPermissionResponse> ask(String title, ToolKind kin
 			ToolCall announce = new ToolCall("tool_call", toolCallId, title, null, kind, ToolCallStatus.PENDING, null, null,
 					null, null, null);
 			ToolCallUpdate toolCall = new ToolCallUpdate(toolCallId, title, kind, ToolCallStatus.PENDING);
-			return agent.sendSessionUpdate(sessionId, announce)
+			return sendUpdate(announce)
 				.then(requestPermission(new RequestPermissionRequest(sessionId, toolCall, options)))
 				.flatMap(response -> {
 					ToolCallStatus status = (response.outcome() instanceof PermissionSelected) ? ToolCallStatus.COMPLETED
 							: ToolCallStatus.FAILED;
-					return agent
-						.sendSessionUpdate(sessionId, new ToolCallUpdateNotification("tool_call_update", toolCallId, null, null, null,
-								status, null, null, null, null, null))
+					return sendUpdate(new ToolCallUpdateNotification("tool_call_update", toolCallId, null, null, null,
+							status, null, null, null, null, null))
 						.thenReturn(response);
 				});
 		});

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

@@ -43,8 +43,9 @@
  * agent's request timeout fails it with a {@link java.util.concurrent.TimeoutException}. When the
  * SDK cancels the handler (after the cancel grace period, or for a {@code $/cancel_request}), it
  * disposes the handler's {@code Mono}: requests still waiting inside it are cancelled, and the
- * client is sent a {@code $/cancel_request} for each. The context does not check that its turn is
- * still active. Its methods may be called from several threads at once.
+ * client is sent a {@code $/cancel_request} for each. Once the prompt has been answered, its
+ * updates are dropped (see {@link #sendUpdate}); its other calls are not checked against the
+ * turn. Its methods may be called from several threads at once.
  *
  * <p>Implementations: the SDK supplies the context handlers receive; implement this interface only
  * for test doubles.
@@ -64,7 +65,10 @@ public interface PromptContext {
 	 * Sends a {@code session/update} notification to the client, carrying one
 	 * {@link AcpSchema.SessionUpdate}: a message or thought chunk, a tool call or its update, a
 	 * plan, and so on, for this prompt's session ({@link #getSessionId()}). The Java client hands a
-	 * turn's updates to its consumers in order, before the prompt's answer. To update another
+	 * turn's updates to its consumers in order, before the prompt's answer. Once the prompt has
+	 * been answered, by the handler or by the SDK when a prompt deadline passed, the SDK's context
+	 * drops further updates (logged at DEBUG) and the {@code Mono} completes empty: ACP requires a
+	 * prompt's updates to precede its answer. To update another
 	 * session, use {@link AcpAsyncAgent#sendSessionUpdate(String, AcpSchema.SessionUpdate)}.
 	 * @param update the update
 	 * @return a {@code Mono} that completes when the notification has been handed to the transport

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

@@ -44,7 +44,8 @@
  * {@link java.util.concurrent.TimeoutException}. When the SDK cancels the handler (after the
  * cancel grace period, or for a {@code $/cancel_request}), it interrupts the handler's thread, and
  * a call blocked at that moment throws {@link java.util.concurrent.CancellationException} and
- * leaves the thread's interrupt flag set. The context does not check that its turn is still active.
+ * leaves the thread's interrupt flag set. Once the prompt has been answered, its updates are
+ * dropped (see {@link #sendUpdate}); its other calls are not checked against the turn.
  *
  * <p>Implementations: the SDK supplies the context handlers receive; implement this interface only
  * for test doubles, and include {@link #async()}.
@@ -65,7 +66,10 @@ public interface SyncPromptContext {
 	 * {@link AcpSchema.SessionUpdate}: a message or thought chunk, a tool call or its update, a
 	 * plan, and so on, for this prompt's session ({@link #getSessionId()}). Returns once the
 	 * notification has been handed to the transport. The Java client hands a turn's updates to its
-	 * consumers in order, before the prompt's answer. To update another session, use
+	 * consumers in order, before the prompt's answer. Once the prompt has been answered, by the
+	 * handler or by the SDK when a prompt deadline passed, the SDK's context drops further
+	 * updates (logged at DEBUG) and returns at once: ACP requires a prompt's updates to precede
+	 * its answer. To update another session, use
 	 * {@link AcpSyncAgent#sendSessionUpdate(String, AcpSchema.SessionUpdate)}.
 	 * @param update the update
 	 */

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

@@ -7,6 +7,7 @@
 import java.time.Duration;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.BooleanSupplier;
 import java.util.function.Function;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicReference;
@@ -414,6 +415,24 @@ public boolean hasActivePrompt(String sessionId) {
 		return activePrompts.isActive(sessionId);
 	}
 
+	/**
+	 * Whether the prompt now running on {@code sessionId} has been answered, by its handler or by
+	 * the session when a prompt deadline passed, as the session sees it later: the SDK's prompt
+	 * contexts ask it before each update, since ACP requires a prompt's updates to precede its
+	 * answer. The answer is about the prompt running at this call; when none is, it is always
+	 * {@code false}.
+	 * @param sessionId the logical ACP session ID
+	 * @return a check, true once that prompt has been answered
+	 */
+	public BooleanSupplier promptAnswered(String sessionId) {
+		Assert.hasText(sessionId, "The sessionId can not be empty");
+		ActivePrompts.Turn turn = activePrompts.current(sessionId);
+		if (turn == null) {
+			return () -> false;
+		}
+		return turn::isAnswered;
+	}
+
 	/**
 	 * Gets one active prompt session ID, if any.
 	 *

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

@@ -1689,9 +1689,9 @@ public RequestPermissionResponse(RequestPermissionOutcome outcome) {
 	 *
 	 * <p>
 	 * The protocol requires an agent to send a turn's updates before it answers the prompt, also
-	 * after a {@code session/cancel}. When the SDK answers a prompt itself, because the cancel
-	 * grace period or the maximum prompt duration passed, a prompt handler that keeps running and
-	 * sends more updates sends them after that answer (see {@link PromptTimeouts}).
+	 * after a {@code session/cancel}. The SDK's prompt contexts drop the updates a handler sends
+	 * once its prompt has been answered, by the handler or by the SDK when the cancel grace period
+	 * or the maximum prompt duration passed (see {@link PromptTimeouts}).
 	 *
 	 * <p>
 	 * The Java client skips, with a warning in the log, a received notification it cannot read or

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

@@ -54,6 +54,9 @@ static final class Turn {
 		/** Completes when the turn ends and releases its session. */
 		private final Sinks.Empty<Void> ended = Sinks.empty();
 
+		/** Set when the turn ends, before its answer is published. */
+		private volatile boolean over;
+
 		/** Runs when a cancel arrives; set by {@link PromptDeadlines} to start the grace period. */
 		private volatile @Nullable Runnable onCancelRequested;
 
@@ -66,6 +69,14 @@ PromptAnswer answer() {
 			return this.answer;
 		}
 
+		/**
+		 * Whether the prompt has been answered: a deadline claimed the answer, or the turn ended,
+		 * which it does just before any answer is published.
+		 */
+		boolean isAnswered() {
+			return this.over || this.answer.isAnswered();
+		}
+
 		/**
 		 * Runs {@code action} when a cancel arrives for this prompt, or now if one already
 		 * has. It may run twice when the two race: it must be idempotent.
@@ -142,6 +153,7 @@ <T> Mono<T> endBeforePublishing(Turn turn, Mono<T> response) {
 	 * @return whether this call released the session
 	 */
 	boolean end(Turn turn, String reason) {
+		turn.over = true;
 		if (this.active.remove(turn.sessionId(), turn)) {
 			logger.debug("Prompt lock released for sessionId={} requestId={} ({})", turn.sessionId(),
 					turn.requestId(), reason);

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

@@ -52,6 +52,11 @@ boolean isCancelling() {
 		return this.state.get() == State.CANCELLING;
 	}
 
+	/** Whether the prompt has been answered, by its handler or by a deadline. */
+	boolean isAnswered() {
+		return this.state.get() == State.ANSWERED;
+	}
+
 	/**
 	 * The handler answered (or failed).
 	 * @return whether its answer is the prompt's answer; false when a deadline answered first