Null-mark the Jetty WebSocket and Streamable HTTP transport packages

Before-start and disconnected state is @nullable (and the WebSocket
transport's server field volatile, read once into a local), as are the
optional session headers and JSON-RPC params the router takes. A
session route scope exposes its always-present id through
boundSessionId(), and the servlet no longer writes a null exception
message into the response.

Author: Mark Pollack

acp-streamable-http-jetty/src/main/java/com/agentclientprotocol/sdk/agent/transport/SseOutboundStream.java

@@ -15,6 +15,7 @@
 import jakarta.servlet.ServletOutputStream;
 import jakarta.servlet.WriteListener;
 import jakarta.servlet.http.HttpServletResponse;
+import org.jspecify.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -55,7 +56,7 @@ final class SseOutboundStream {
 	private final ArrayDeque<String> mailbox = new ArrayDeque<>();
 
 	/** Guarded by {@code this}. */
-	private SseSubscriber current;
+	private @Nullable SseSubscriber current;
 
 	/** Guarded by {@code this}. */
 	private boolean closed;

acp-streamable-http-jetty/src/main/java/com/agentclientprotocol/sdk/agent/transport/StreamableHttpAcpAgentTransport.java

@@ -24,6 +24,7 @@
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.server.ServerConnector;
 import org.eclipse.jetty.websocket.server.WebSocketUpgradeHandler;
+import org.jspecify.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
@@ -94,9 +95,9 @@ public class StreamableHttpAcpAgentTransport {
 
 	private final Sinks.One<Void> terminationSink = Sinks.one();
 
-	private volatile Server server;
+	private volatile @Nullable Server server;
 
-	private volatile ServerConnector connector;
+	private volatile @Nullable ServerConnector connector;
 
 	/**
 	 * Creates a new Streamable HTTP listener on the default ACP path.

acp-streamable-http-jetty/src/main/java/com/agentclientprotocol/sdk/agent/transport/StreamableHttpAcpServlet.java

@@ -30,6 +30,7 @@
 import jakarta.servlet.http.HttpServlet;
 import jakarta.servlet.http.HttpServletRequest;
 import jakarta.servlet.http.HttpServletResponse;
+import org.jspecify.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.Disposable;
@@ -89,9 +90,9 @@ public class StreamableHttpAcpServlet extends HttpServlet {
 
 	private final transient AtomicBoolean closing = new AtomicBoolean(false);
 
-	private transient volatile Scheduler keepAliveScheduler;
+	private transient volatile @Nullable Scheduler keepAliveScheduler;
 
-	private transient volatile Disposable keepAliveTask;
+	private transient volatile @Nullable Disposable keepAliveTask;
 
 	/**
 	 * Creates a servlet with the default limits.
@@ -428,10 +429,12 @@ private Optional<String> header(HttpServletRequest request, String name) {
 		return Optional.ofNullable(request.getHeader(name)).filter(value -> !value.isBlank());
 	}
 
-	private void writeText(HttpServletResponse response, int status, String body) throws IOException {
+	private void writeText(HttpServletResponse response, int status, @Nullable String body) throws IOException {
 		response.setStatus(status);
 		response.setContentType("text/plain");
-		response.getWriter().write(body);
+		if (body != null) {
+			response.getWriter().write(body);
+		}
 	}
 
 }

acp-streamable-http-jetty/src/main/java/com/agentclientprotocol/sdk/agent/transport/StreamableHttpConnection.java

@@ -24,6 +24,7 @@
 import jakarta.servlet.AsyncContext;
 import jakarta.servlet.http.HttpServletRequest;
 import jakarta.servlet.http.HttpServletResponse;
+import org.jspecify.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
@@ -66,7 +67,7 @@ final class StreamableHttpConnection {
 
 	private final AtomicBoolean initialized = new AtomicBoolean(false);
 
-	private volatile Object initializeRequestId;
+	private volatile @Nullable Object initializeRequestId;
 
 	private final AcpJsonMapper jsonMapper;
 
@@ -105,7 +106,7 @@ Mono<JSONRPCMessage> initialize(AcpSchema.JSONRPCRequest request) {
 		return initializeResponse.asMono().doOnSuccess(ignored -> initialized.set(true));
 	}
 
-	void acceptClientPost(JSONRPCMessage message, String sessionHeader) {
+	void acceptClientPost(JSONRPCMessage message, @Nullable String sessionHeader) {
 		if (message instanceof AcpSchema.JSONRPCResponse response) {
 			validateClientResponseScope(response, sessionHeader);
 			connection.acceptInbound(message);
@@ -114,7 +115,7 @@ void acceptClientPost(JSONRPCMessage message, String sessionHeader) {
 
 		ResolvedInboundRoute resolved = routing.resolveInboundRoute(message, sessionHeader);
 		if (resolved.requestScope().isSession()) {
-			prepareSessionForInbound(resolved.requestScope().sessionId(), resolved.requestRoute());
+			prepareSessionForInbound(resolved.requestScope().boundSessionId(), resolved.requestRoute());
 		}
 		if (message instanceof AcpSchema.JSONRPCRequest request && request.id() != null
 				&& resolved.requestRoute() != null) {
@@ -123,12 +124,12 @@ void acceptClientPost(JSONRPCMessage message, String sessionHeader) {
 		connection.acceptInbound(message);
 	}
 
-	void openStream(HttpServletRequest request, HttpServletResponse response, String sessionId)
+	void openStream(HttpServletRequest request, HttpServletResponse response, @Nullable String sessionId)
 			throws IOException {
 		RouteScope scope = sessionId == null ? RouteScope.connection() : RouteScope.session(sessionId);
 		SseOutboundStream stream;
 		if (scope.isSession()) {
-			stream = openSessionStream(scope.sessionId());
+			stream = openSessionStream(scope.boundSessionId());
 		}
 		else {
 			stream = connectionStream;
@@ -139,7 +140,7 @@ void openStream(HttpServletRequest request, HttpServletResponse response, String
 		response.setHeader("Cache-Control", "no-cache");
 		response.setHeader(HEADER_CONNECTION_ID, id);
 		if (scope.isSession()) {
-			response.setHeader(HEADER_SESSION_ID, scope.sessionId());
+			response.setHeader(HEADER_SESSION_ID, scope.boundSessionId());
 		}
 		AsyncContext asyncContext = request.startAsync();
 		asyncContext.setTimeout(0);
@@ -169,7 +170,7 @@ private void routeAgentMessage(JSONRPCMessage message) {
 			RouteScope scope = resolveAgentOutboundScope(message);
 			String payload = jsonMapper.writeValueAsString(message);
 			if (scope.isSession()) {
-				sessionStream(scope.sessionId()).push(payload);
+				sessionStream(scope.boundSessionId()).push(payload);
 			}
 			else {
 				connectionStream.push(payload);
@@ -197,10 +198,10 @@ private RouteScope resolveAgentOutboundScope(JSONRPCMessage message) {
 			}
 			if (route.kind() == RequestKind.SESSION_LOAD) {
 				if (response.error() == null) {
-					markSessionKnown(route.requestScope().sessionId());
+					markSessionKnown(route.requestScope().boundSessionId());
 				}
 				else {
-					discardProvisionalSession(route.requestScope().sessionId());
+					discardProvisionalSession(route.requestScope().boundSessionId());
 				}
 			}
 			return route.responseScope();
@@ -229,7 +230,7 @@ else if (message instanceof AcpSchema.JSONRPCNotification notification) {
 		return scope;
 	}
 
-	private void prepareSessionForInbound(String sessionId, ClientRequestRoute route) {
+	private void prepareSessionForInbound(String sessionId, @Nullable ClientRequestRoute route) {
 		SessionState current = sessions.get(sessionId);
 		if (route != null && route.kind() == RequestKind.SESSION_LOAD) {
 			if (current == null) {
@@ -243,7 +244,7 @@ private void prepareSessionForInbound(String sessionId, ClientRequestRoute route
 		}
 	}
 
-	private void validateClientResponseScope(AcpSchema.JSONRPCResponse response, String sessionHeader) {
+	private void validateClientResponseScope(AcpSchema.JSONRPCResponse response, @Nullable String sessionHeader) {
 		RouteScope expected = agentRequestRoutes.get(response.id());
 		if (expected == null) {
 			logger.warn("Client posted response for unknown agent request id {}", response.id());

acp-streamable-http-jetty/src/main/java/com/agentclientprotocol/sdk/agent/transport/StreamableHttpRouting.java

@@ -12,6 +12,7 @@
 import com.agentclientprotocol.sdk.json.TypeRef;
 import com.agentclientprotocol.sdk.spec.AcpSchema;
 import com.agentclientprotocol.sdk.spec.AcpSchema.JSONRPCMessage;
+import org.jspecify.annotations.Nullable;
 
 /**
  * Routing rules of the Streamable HTTP transport: which stream (connection or session)
@@ -63,7 +64,7 @@ enum SessionState {
 
 	}
 
-	record RouteScope(ScopeKind kind, String sessionId) {
+	record RouteScope(ScopeKind kind, @Nullable String sessionId) {
 
 		static RouteScope connection() {
 			return new RouteScope(ScopeKind.CONNECTION, null);
@@ -77,12 +78,24 @@ boolean isSession() {
 			return kind == ScopeKind.SESSION;
 		}
 
+		/**
+		 * The session id this scope routes to. Only session scopes carry one, and every
+		 * caller has already established that the scope is one.
+		 */
+		String boundSessionId() {
+			if (sessionId == null) {
+				throw new IllegalStateException("A " + kind + " route scope has no session id");
+			}
+			return sessionId;
+		}
+
 	}
 
 	record ClientRequestRoute(RequestKind kind, RouteScope requestScope, RouteScope responseScope) {
 	}
 
-	record ResolvedInboundRoute(JSONRPCMessage message, RouteScope requestScope, ClientRequestRoute requestRoute) {
+	record ResolvedInboundRoute(JSONRPCMessage message, RouteScope requestScope,
+			@Nullable ClientRequestRoute requestRoute) {
 	}
 
 	static boolean isInitialize(JSONRPCMessage message) {
@@ -113,9 +126,9 @@ RouteScope resolveAgentRequestOrNotificationScope(String method, Object params)
 		}
 	}
 
-	ResolvedInboundRoute resolveInboundRoute(JSONRPCMessage message, String sessionHeader) {
+	ResolvedInboundRoute resolveInboundRoute(JSONRPCMessage message, @Nullable String sessionHeader) {
 		String method;
-		Object params;
+		@Nullable Object params;
 		if (message instanceof AcpSchema.JSONRPCRequest request) {
 			method = request.method();
 			params = request.params();
@@ -178,14 +191,14 @@ else if (message instanceof AcpSchema.JSONRPCNotification notification) {
 	}
 
 	/** Like {@link #requireSessionScope} but tolerates a missing header; a conflicting one is still rejected. */
-	RouteScope sessionScopeFromParams(String method, Object params, String sessionHeader) {
+	RouteScope sessionScopeFromParams(String method, @Nullable Object params, @Nullable String sessionHeader) {
 		if (sessionHeader == null) {
 			return RouteScope.session(requireSessionId(params, method));
 		}
 		return requireSessionScope(method, params, sessionHeader);
 	}
 
-	RouteScope requireSessionScope(String method, Object params, String sessionHeader) {
+	RouteScope requireSessionScope(String method, @Nullable Object params, @Nullable String sessionHeader) {
 		String sessionId = requireSessionId(params, method);
 		if (sessionHeader == null) {
 			throw new AcpConnectionException(
@@ -198,7 +211,7 @@ RouteScope requireSessionScope(String method, Object params, String sessionHeade
 		return RouteScope.session(sessionId);
 	}
 
-	Optional<String> extractSessionId(Object params) {
+	Optional<String> extractSessionId(@Nullable Object params) {
 		if (params == null) {
 			return Optional.empty();
 		}
@@ -207,7 +220,7 @@ Optional<String> extractSessionId(Object params) {
 		return sessionId == null ? Optional.empty() : Optional.of(sessionId.toString());
 	}
 
-	String requireSessionId(Object params, String method) {
+	String requireSessionId(@Nullable Object params, String method) {
 		return extractSessionId(params)
 			.filter(sessionId -> !sessionId.isBlank())
 			.orElseThrow(() -> new AcpConnectionException("Missing sessionId for method " + method));

acp-streamable-http-jetty/src/main/java/com/agentclientprotocol/sdk/agent/transport/StreamableHttpWebSocketConnection.java

@@ -23,6 +23,7 @@
 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage;
 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketOpen;
 import org.eclipse.jetty.websocket.api.annotations.WebSocket;
+import org.jspecify.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
@@ -50,7 +51,8 @@ final class StreamableHttpWebSocketConnection {
 
 	private final SerializedWebSocketSender outboundSender = new SerializedWebSocketSender();
 
-	private volatile Session session;
+	/** The client's WebSocket session; null until {@link #open}. */
+	private volatile @Nullable Session session;
 
 	private final AcpJsonMapper jsonMapper;
 

acp-streamable-http-jetty/src/main/java/com/agentclientprotocol/sdk/agent/transport/package-info.java

@@ -0,0 +1,16 @@
+/*
+ * Copyright 2025-2026 the original author or authors.
+ */
+
+/**
+ * The Jetty Streamable HTTP agent transport (acp-streamable-http-jetty); this module's part of the package shared with acp-core.
+ *
+ * <p>
+ * The package is null-marked: every type is non-null unless annotated
+ * {@link org.jspecify.annotations.Nullable @Nullable}.
+ * </p>
+ */
+@NullMarked
+package com.agentclientprotocol.sdk.agent.transport;
+
+import org.jspecify.annotations.NullMarked;

acp-websocket-jetty/src/main/java/com/agentclientprotocol/sdk/agent/transport/WebSocketAcpAgentTransport.java

@@ -26,6 +26,7 @@
 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketOpen;
 import org.eclipse.jetty.websocket.api.annotations.WebSocket;
 import org.eclipse.jetty.websocket.server.WebSocketUpgradeHandler;
+import org.jspecify.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
@@ -85,7 +86,8 @@ public class WebSocketAcpAgentTransport implements AcpAgentTransport {
 
 	private final String path;
 
-	private Server server;
+	/** The embedded Jetty server; null until {@link #start} creates it. */
+	private volatile @Nullable Server server;
 
 	private final Sinks.Many<JSONRPCMessage> inboundSink;
 
@@ -103,7 +105,8 @@ public class WebSocketAcpAgentTransport implements AcpAgentTransport {
 
 	private Consumer<Throwable> exceptionHandler = t -> logger.error("Transport error", t);
 
-	private volatile Session clientSession;
+	/** The connected client's session; null while no client is connected. */
+	private volatile @Nullable Session clientSession;
 
 	private Duration idleTimeout = Duration.ofMinutes(30);
 
@@ -173,7 +176,8 @@ public Mono<Void> start(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> han
 			handleIncomingMessages(handler);
 
 			// Create and configure Jetty server
-			server = new Server();
+			Server server = new Server();
+			this.server = server;
 			ServerConnector connector = new ServerConnector(server);
 			connector.setPort(port);
 			server.addConnector(connector);
@@ -278,6 +282,7 @@ public Mono<Void> closeGracefully() {
 			if (currentSession != null && currentSession.isOpen()) {
 				currentSession.close();
 			}
+			Server server = this.server;
 			if (server != null) {
 				server.stop();
 			}

acp-websocket-jetty/src/main/java/com/agentclientprotocol/sdk/agent/transport/package-info.java

@@ -0,0 +1,16 @@
+/*
+ * Copyright 2025-2026 the original author or authors.
+ */
+
+/**
+ * The Jetty WebSocket agent transport (acp-websocket-jetty); this module's part of the package shared with acp-core.
+ *
+ * <p>
+ * The package is null-marked: every type is non-null unless annotated
+ * {@link org.jspecify.annotations.Nullable @Nullable}.
+ * </p>
+ */
+@NullMarked
+package com.agentclientprotocol.sdk.agent.transport;
+
+import org.jspecify.annotations.NullMarked;