/*
 * Copyright 2022-2026 Revetware LLC.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.soklet;

import com.google.errorprone.annotations.CanIgnoreReturnValue;
import com.google.errorprone.annotations.CheckReturnValue;
import com.soklet.internal.mcp.protocol.McpServerRuntimeBridge.SimulationSession;
import com.soklet.internal.microhttp.StreamLifecycleCoordinator;

import com.soklet.SseRequestResult.HandshakeAccepted;
import com.soklet.SseRequestResult.HandshakeRejected;
import com.soklet.annotation.SseEventSource;
import com.soklet.exception.IllegalRequestException;
import com.soklet.internal.spring.LinkedCaseInsensitiveMap;
import com.soklet.internal.streaming.ManagedResponseStream;
import com.soklet.internal.streaming.ManagedSseLifecycle;
import com.soklet.internal.streaming.PublisherResponseStream;
import org.jspecify.annotations.NonNull;
import org.jspecify.annotations.Nullable;

import javax.annotation.concurrent.NotThreadSafe;
import javax.annotation.concurrent.ThreadSafe;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.Reader;
import java.lang.reflect.InvocationTargetException;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.Charset;
import java.nio.charset.CharsetEncoder;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.CoderResult;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.LinkedHashSet;
import java.util.HashSet;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import static com.soklet.internal.ObjectIdentity.sameInstance;
import static com.soklet.Utilities.emptyByteArray;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;

/**
 * Owns one ready-to-terminal Soklet lifecycle and its configured HTTP, SSE,
 * and MCP transports. {@link #run(SokletConfig)} returns
 * after all configured transports reach readiness; each instance is one-shot.
 * Every run registers a JVM shutdown hook by default. An embedding host can
 * disable only that registration through
 * {@link SokletConfig.Builder#jvmShutdownHookEnabled(Boolean)}. Bounded application
 * cleanup, terminal reporting, and process-registration release remain owned
 * by the instance until finalization finishes, even when nobody waits.
 * <p>
 * <pre>{@code
 * SokletConfig config = SokletConfig.withHttpServer(HttpServer.fromPort(8080)).build();
 * Soklet soklet = Soklet.run(config);
 * ShutdownResult result = soklet.awaitShutdown();
 * }</pre>
 * To stop programmatically, call {@link #requestShutdown()} and then await
 * completion from an independent owning thread. An embedding host must finish
 * Soklet shutdown before closing shared databases, telemetry, or logging.
 * JVM hooks execute concurrently and do not establish resource ordering.
 * With the automatic hook disabled, the host must prevent new runs once
 * termination begins and request and synchronously await shutdown before its
 * own shutdown hook returns. Any deregistration or delay belongs before that
 * shutdown request; shared dependencies remain available until shutdown succeeds.
 * <p>
 * Required application cleanup belongs in bounded {@link ShutdownCleanup},
 * rather than post-wait code or observational lifecycle callbacks: during JVM
 * shutdown the JVM need not let those callers finish. Cleanup is eligible only
 * after complete core shutdown; a cleanup timeout does not forcibly stop its code.
 * Running built-in listeners and the lifecycle owner retain JVM liveness until
 * shutdown and finalization; custom transports own their listener-thread policy.
 * <p>
 * Soklet also offers an off-network {@link Simulator} through
 * {@link SokletSimulator}, useful for integration testing.
 * <p>
 * Given a <em>Resource Method</em>...
 * <pre>{@code public class HelloResource {
 *   @GET("/hello")
 *   public String hello(@QueryParameter String name) {
 *     return String.format("Hello, %s", name);
 *   }
 * }}</pre>
 * ...we might test it like this:
 * <pre>{@code @Test
 * public void integrationTest() {
 *   // Instead of running on a real HTTP server that listens on a port,
 *   // a non-network Simulator is provided against which you can
 *   // issue requests and receive responses.
 *   SokletSimulator.run(SimulatorConfig.builder().httpServer().build(), simulator -> {
 *     // Construct a request
 *     Request request = Request.withPath(HttpMethod.GET, "/hello")
 *       .queryParameters(Map.of("name", List.of("Mark")))
 *       .build();
 *
 *     // Perform the request and get a handle to the response
 *     HttpRequestResult result = simulator.performHttpRequest(request);
 *     MarshaledResponse marshaledResponse = result.getMarshaledResponse();
 *
 *     // Verify status code
 *     Integer expectedCode = 200;
 *     Integer actualCode = marshaledResponse.getStatusCode();
 *     assertEquals(expectedCode, actualCode, "Bad status code");
 *
 *     // Verify response body
 *     marshaledResponse.getBody().ifPresentOrElse(body -> {
 *       String expectedBody = "Hello, Mark";
 *       byte[] bytes = ((MarshaledResponseBody.Bytes) body).getBytes();
 *       String actualBody = new String(bytes, StandardCharsets.UTF_8);
 *       assertEquals(expectedBody, actualBody, "Bad response body");
 *     }, () -> {
 *       Assertions.fail("No response body");
 *     });
 *   });
 * }}</pre>
 * <p>
 * The {@link Simulator} also supports Server-Sent Events.
 * <p>
 * Integration testing documentation is available at <a href="https://www.soklet.com/docs/testing">https://www.soklet.com/docs/testing</a>.
 *
 * @author <a href="https://www.revetkn.com">Mark Allen</a>
 */
@ThreadSafe
public final class Soklet {
        /**
         * Application work whose return or failure ends one running Soklet scope.
         * The action runs once, on the calling thread after readiness, without a
         * lifecycle handle. Checked failures retain their application-selected type.
         * The action itself has no imposed timeout and is not interrupted by Soklet
         * when a transport stops or JVM shutdown begins.
         *
         * @param <E> throwable type selected by the action
         */
        @NotThreadSafe
        @FunctionalInterface
        public interface RunAction<E extends @NonNull Throwable> {
                /**
                 * Performs work while the enclosing caller owns the running application.
                 *
                 * @throws E if application work fails
                 */
                void run() throws E;
        }

        @NonNull
        private static final Map<@NonNull String, @NonNull List<@NonNull String>> DEFAULT_ACCEPTED_HANDSHAKE_HEADERS;

        static {
                // Generally speaking, we always want these headers for SSE streaming responses.
                // Users can override if they think necessary
                LinkedCaseInsensitiveMap<List<String>> defaultAcceptedHandshakeHeaders = new LinkedCaseInsensitiveMap<>(4);
                defaultAcceptedHandshakeHeaders.put("Content-Type", List.of("text/event-stream; charset=UTF-8"));
                defaultAcceptedHandshakeHeaders.put("Cache-Control", List.of("no-cache", "no-transform"));
                defaultAcceptedHandshakeHeaders.put("Connection", List.of("keep-alive"));
                defaultAcceptedHandshakeHeaders.put("X-Accel-Buffering", List.of("no"));

                DEFAULT_ACCEPTED_HANDSHAKE_HEADERS = Collections.unmodifiableMap(defaultAcceptedHandshakeHeaders);
        }

        /**
         * Constructs an internal lifecycle for transport-isolated simulation and
         * low-level lifecycle composition. Claims transport identities immediately.
         *
         * @param sokletConfig configuration that drives the Soklet system
         * @return a Soklet instance
         * @throws TransportOwnershipException if a configured transport identity
         * has already been claimed by another lifecycle
         */
        @NonNull
        static Soklet fromConfig(@NonNull SokletConfig sokletConfig) {
                requireNonNull(sokletConfig);
                return new Soklet(sokletConfig, LifecycleRuntimeServices.system(),
                                ignored -> { });
        }

        @NonNull
        static Soklet fromConfig(@NonNull SokletConfig sokletConfig,
                        @NonNull LifecycleRuntimeServices services,
                        @NonNull Consumer<@NonNull InternalLifecycleCoreSnapshot> coreSnapshotPublisher) {
                return new Soklet(requireNonNull(sokletConfig), requireNonNull(services),
                                requireNonNull(coreSnapshotPublisher));
        }

        @NonNull
        private final SokletConfig sokletConfig;
        @NonNull
        private final ReentrantLock lock;
        @NonNull
        private final CountDownLatch awaitShutdownLatch;
        @NonNull
        private final AtomicReference<@NonNull CountDownLatch>
                        awaitShutdownLatchReference;
        @NonNull
        private final SokletFrameworkSetup frameworkSetup;
        @NonNull
        private final SokletDirectLifecycle directLifecycle;
        @NonNull
        private final CompletionStage<@NonNull ShutdownResult> coreAwaitStage;
        private volatile @Nullable SokletRunCompletion runCompletion;

        /**
         * Creates a Soklet instance with the given configuration.
         *
         * @param sokletConfig configuration that drives the Soklet system
         */
        private Soklet(@NonNull SokletConfig sokletConfig,
                        @NonNull LifecycleRuntimeServices services,
                        @NonNull Consumer<@NonNull InternalLifecycleCoreSnapshot> coreSnapshotPublisher) {
                requireNonNull(sokletConfig);

                this.sokletConfig = sokletConfig;
                this.lock = new ReentrantLock();
                this.awaitShutdownLatch = new CountDownLatch(1);
                this.awaitShutdownLatchReference =
                                new AtomicReference<>(this.awaitShutdownLatch);
                this.frameworkSetup = new SokletFrameworkSetup(sokletConfig);
                this.directLifecycle = new SokletDirectLifecycle(this, sokletConfig,
                                this.frameworkSetup, requireNonNull(services),
                                requireNonNull(coreSnapshotPublisher));
                this.coreAwaitStage = this.directLifecycle.shutdownStage().thenApply(result -> {
                        this.directLifecycle.throwIfUnsuccessfulShutdown(result);
                        return result;
                });
        }

        /**
         * Starts one lifecycle, returning only after every configured transport has
         * reached readiness. By default owns a JVM shutdown hook until full finalization;
         * {@link SokletConfig.Builder#jvmShutdownHookEnabled(Boolean)} may disable only
         * that automatic registration. Programmatic shutdown, cleanup, and lifecycle
         * owner liveness remain unchanged. The host then owns signal-driven shutdown
         * and must request and synchronously await it before its own hook returns.
         * Startup failure or cancellation throws
         * synchronously after attempted teardown; interruption before readiness
         * requests cancellation and the caller's interrupt status is restored.
         * Cancellation competes atomically with readiness. If readiness wins, the
         * ready handle is returned with the interrupt status restored, and that
         * startup interruption does not request shutdown of the running lifecycle.
         *
         * @param sokletConfig configuration with unclaimed transport identities
         * @return the ready lifecycle handle
         * @throws TransportOwnershipException if a configured transport is already claimed
         * @throws SokletStartupException if readiness is not reached
         */
        @NonNull
        public static Soklet run(@NonNull SokletConfig sokletConfig) {
                return SokletProcessLifecycle.launch(requireNonNull(sokletConfig), null);
        }

        /**
         * Starts a lifecycle with bounded application cleanup eligible after complete
         * core shutdown. All arguments are validated before transports are claimed.
         * When enabled, the JVM hook joins bounded core shutdown, cleanup, and reporting.
         * Both await methods additionally observe process-registration release.
         *
         * @param sokletConfig configuration with unclaimed transport identities
         * @param cleanup bounded synchronous application cleanup
         * @return the ready lifecycle handle
         * @throws TransportOwnershipException if a configured transport is already claimed
         * @throws SokletStartupException if readiness is not reached
         */
        @NonNull
        public static Soklet run(@NonNull SokletConfig sokletConfig,
                        @NonNull ShutdownCleanup cleanup) {
                return SokletProcessLifecycle.launch(requireNonNull(sokletConfig),
                                requireNonNull(cleanup));
        }

        /**
         * Starts Soklet, invokes the action after readiness, and requests shutdown
         * when the action returns or throws. Joins full finalization before returning
         * the immutable terminal result or propagating a failure. The action runs
         * once on the calling thread; startup failure prevents its invocation.
         * All arguments are validated before transports are claimed.
         * The configuration's {@link SokletConfig#isJvmShutdownHookEnabled()} option
         * applies to scoped execution as well; disabling the automatic hook does not
         * disable teardown when the action returns or throws.
         * <p>
         * A callback failure remains the exact primary exception or Error. A
         * concurrent finalization failure is suppressed on it using ordinary Java
         * suppression semantics, after finalization has finished. Without a callback
         * failure, finalization uses the same failure classification as
         * {@link #awaitShutdown()}. A complete forced shutdown can return normally.
         * <p>
         * Finalization does not abandon shutdown when the caller is interrupted.
         * An interrupt flag left set by the action is temporarily cleared for the
         * join, and it and interruptions received while joining are restored before
         * returning or throwing. An InterruptedException thrown by the action is
         * propagated as that action's original failure.
         * <p>
         * The action has no imposed timeout. An external shutdown does not terminate
         * it; the scoped invocation still needs the action to return or throw.
         *
         * @param sokletConfig configuration with unclaimed transport identities
         * @param action work defining the scope's lifetime
         * @param <E> throwable type selected by the action
         * @return the immutable successful terminal result; it may be ignored
         * @throws E if the action fails, after attempted shutdown and full finalization
         * @throws SokletStartupException if startup does not reach readiness
         * @throws SokletLifecycleException if finalization fails without an action failure
         */
        @NonNull
        public static <E extends @NonNull Throwable> ShutdownResult run(
                        @NonNull SokletConfig sokletConfig,
                        @NonNull RunAction<@NonNull E> action) throws E {
                return runScoped(sokletConfig, null, action);
        }

        /**
         * Runs one action-defined scope with bounded application cleanup. Shares the
         * execution, error, and interruption semantics of
         * {@link #run(SokletConfig, RunAction)}. Cleanup is eligible only after
         * complete core shutdown, including when the action failed; incomplete core
         * shutdown skips cleanup. All arguments are validated before transport claims.
         *
         * @param sokletConfig configuration with unclaimed transport identities
         * @param cleanup bounded synchronous application cleanup
         * @param action work defining the scope's lifetime
         * @param <E> throwable type selected by the action
         * @return the immutable successful terminal result; it may be ignored
         * @throws E if the action fails, after attempted shutdown and full finalization
         * @throws SokletStartupException if startup does not reach readiness
         * @throws SokletLifecycleException if finalization fails without an action failure
         */
        @NonNull
        public static <E extends @NonNull Throwable> ShutdownResult run(
                        @NonNull SokletConfig sokletConfig,
                        @NonNull ShutdownCleanup cleanup,
                        @NonNull RunAction<@NonNull E> action) throws E {
                return runScoped(sokletConfig, requireNonNull(cleanup), action);
        }

        @NonNull
        private static <E extends @NonNull Throwable> ShutdownResult runScoped(
                        @NonNull SokletConfig sokletConfig, @Nullable ShutdownCleanup cleanup,
                        @NonNull RunAction<E> action) throws E {
                SokletConfig exactConfig = requireNonNull(sokletConfig);
                RunAction<E> exactAction = requireNonNull(action);
                Soklet soklet = SokletProcessLifecycle.launch(exactConfig, cleanup);
                Throwable actionFailure = null;
                try {
                        exactAction.run();
                } catch (Throwable failure) {
                        actionFailure = failure;
                }

                boolean interrupted = Thread.interrupted();
                Throwable requestFailure = null;
                Throwable finalizationFailure = null;
                ShutdownResult result = null;
                try {
                        try {
                                soklet.requestShutdown();
                        } catch (Throwable failure) {
                                requestFailure = failure;
                        }
                        for (;;) {
                                try {
                                        result = soklet.awaitShutdown();
                                        break;
                                } catch (InterruptedException interruption) {
                                        interrupted = true;
                                } catch (Throwable failure) {
                                        finalizationFailure = failure;
                                        break;
                                }
                        }
                } finally {
                        if (interrupted)
                                Thread.currentThread().interrupt();
                }

                // Never acquire an application exception's monitor until the lifecycle
                // has finished, and never combine failures while holding lifecycle locks.
                if (finalizationFailure == null)
                        finalizationFailure = requestFailure;
                else if (requestFailure != null && !sameInstance(finalizationFailure, requestFailure))
                        finalizationFailure.addSuppressed(requestFailure);
                if (actionFailure != null) {
                        if (finalizationFailure != null && !sameInstance(actionFailure, finalizationFailure))
                                actionFailure.addSuppressed(finalizationFailure);
                        Soklet.<E>throwRunFailure(actionFailure);
                }
                if (finalizationFailure != null)
                        Soklet.<E>throwRunFailure(finalizationFailure);
                return requireNonNull(result);
        }

        @SuppressWarnings("unchecked")
        private static <E extends @NonNull Throwable> void throwRunFailure(
                        @NonNull Throwable failure) throws E {
                throw (E) requireNonNull(failure);
        }

        /**
         * Requests shutdown promptly without waiting. Repeated requests are safe.
         * Handlers and callbacks may request shutdown but must return without waiting
         * for work whose termination depends on their own return.
         */
        public void requestShutdown() {
                this.directLifecycle.requestShutdownIntent();
        }

        /**
         * Claims this one-shot lifecycle's sole start attempt and returns only after
         * every configured transport is globally ready.
         * If the calling thread is already interrupted, startup is canceled before
         * readiness and fails with {@link SokletStartupException}; its lifecycle
         * result records {@link StartupDisposition#CANCELED}.
         *
         * @throws SokletStartupException if startup ends before readiness, carrying
         * the exact immutable lifecycle result
         * @throws IllegalStateException if another start call already claimed this
         * lifecycle attempt
         */
        void start() {
                this.directLifecycle.start();
        }

        /**
         * Publishes Soklet-wide shutdown intent promptly and returns the one cached,
         * read-only completion stage for this lifecycle attempt.
         * <p>
         * This method does not install or own process hooks, operating-system signal
         * handlers, or standard-input listeners; direct lifecycle callers own those
         * integrations.
         * <p>
         * Handlers, lifecycle/metric callbacks and transport-owned work may request
         * shutdown here, but must return without joining this stage or waiting on a
         * future derived from it. Shutdown can be waiting for that same work to exit;
         * a self-wait can consume the shutdown budget and publish an incomplete result.
         * Wait for completion from the owning process or another independent thread.
         *
         * @return cached shutdown-result stage
         */
        @NonNull
        CompletionStage<@NonNull ShutdownResult> shutdown() {
                return this.directLifecycle.shutdown();
        }

        /**
         * Waits for full shutdown, bounded application cleanup, terminal reporting,
         * and process-registration release, without requesting shutdown. This wait
         * and {@link #awaitShutdownAsync()} expose the same immutable result or underlying
         * lifecycle failure; the asynchronous stage wraps failures in
         * {@link java.util.concurrent.CompletionException}.
         *
         * @return the exact immutable terminal result
         * @throws InterruptedException if this wait is interrupted; the lifecycle continues
         * @throws SokletLifecycleException if the lifecycle or eligible cleanup fails
         * @throws IllegalStateException if tracked lifecycle work attempts to join itself
         */
        @NonNull
        @CanIgnoreReturnValue
        public ShutdownResult awaitShutdown() throws InterruptedException {
                SokletRunCompletion completion = this.runCompletion;
                if (completion != null)
                        return completion.await();
                ShutdownResult result = awaitCoreShutdown();
                this.directLifecycle.throwIfUnsuccessfulShutdown(result);
                return result;
        }

        /**
         * Observes the same full terminal outcome as {@link #awaitShutdown()} without
         * blocking or requesting shutdown. Returns one cached read-only stage; no
         * waiting thread is created by this method. Lifecycle failures complete it
         * exceptionally, with the underlying lifecycle failure wrapped in
         * {@link java.util.concurrent.CompletionException}. Mutating a future derived
         * from it cannot alter the lifecycle.
         * Callbacks must not join this stage from work that shutdown must drain.
         * The daemon completion handoff does not keep the JVM alive, including during
         * normal programmatic shutdown after {@code main} returns. A host composing
         * asynchronous cleanup must retain process liveness until its dependent stages
         * finish. Put required standalone application cleanup in {@link ShutdownCleanup}.
         *
         * @return cached full-finalization completion stage
         */
        @NonNull
        @CheckReturnValue
        public CompletionStage<@NonNull ShutdownResult> awaitShutdownAsync() {
                SokletRunCompletion completion = this.runCompletion;
                return completion == null ? this.coreAwaitStage : completion.async();
        }

        @NonNull
        ShutdownResult awaitCoreShutdown() throws InterruptedException {
                return this.directLifecycle.awaitPublicCompletion();
        }

        void attachRunCompletion(@NonNull SokletRunCompletion completion) {
                this.runCompletion = requireNonNull(completion);
        }

        /**
         * Nonpublic "informal" implementation of {@link com.soklet.HttpServer.RequestHandler} so Soklet does not need to expose {@code handleRequest} publicly.
         * Reasoning: users of this library should never call {@code handleRequest} directly - it should only be invoked in response to events
         * provided by a {@link HttpServer} or {@link SseServer} implementation.
         */
        void handleRequest(@NonNull Request request,
                                                                                                                 @NonNull ServerType serverType,
                                                                                                                 @NonNull Consumer<@NonNull HttpRequestResult> requestResultConsumer) {
                try (LifecycleExecutionContext.Scope ignored =
                                         this.directLifecycle.enterExecution()) {
                        handleRequestWithinLifecycle(request, serverType,
                                        requestResultConsumer);
                }
        }

        private void handleRequestWithinLifecycle(@NonNull Request request,
                                                                                                                 @NonNull ServerType serverType,
                                                                                                                 @NonNull Consumer<@NonNull HttpRequestResult> requestResultConsumer) {
                requireNonNull(request);
                requireNonNull(serverType);
                requireNonNull(requestResultConsumer);

                long processingStartedNanos = System.nanoTime();

                SokletConfig sokletConfig = getSokletConfig();
                ResourceMethodResolver resourceMethodResolver = sokletConfig.getResourceMethodResolver();
                ResponseMarshaler responseMarshaler = sokletConfig.getResponseMarshaler();
                ResponseFinalizationToken postProcessingToken = newPostProcessingToken();
                LifecycleObserver lifecycleObserver = sokletConfig.getAggregateLifecycleObserver();
                RequestInterceptor requestInterceptor = sokletConfig.getRequestInterceptor();
                MetricsCollector metricsCollector = sokletConfig.getMetricsCollector();

                // Holders to permit mutable effectively-final variables
                AtomicReference<@Nullable MarshaledResponse> marshaledResponseHolder = new AtomicReference<>();
                AtomicReference<@Nullable Throwable> resourceMethodResolutionExceptionHolder = new AtomicReference<>();
                AtomicReference<@NonNull Request> requestHolder = new AtomicReference<>(request);
                // The handler's effective request may be replaced by either interceptor hook. Paired
                // lifecycle/metrics callbacks retain the original per-dispatch identity, as do HTTP
                // stream handles, so observers never have to correlate caller-controlled request IDs.
                AtomicReference<@Nullable ResourceMethod> resourceMethodHolder = new AtomicReference<>();
                AtomicReference<@Nullable HttpRequestResult> requestResultHolder = new AtomicReference<>();
                AtomicReference<@Nullable Throwable> requestHandlingFailureHolder = new AtomicReference<>();
                Consumer<MarshaledResponse> captureHeadRepresentation = representation -> {
                        if (serverType != ServerType.HTTP || requestHolder.get().getHttpMethod() != HttpMethod.HEAD)
                                return;
                        MarshaledResponseBody compressionBody = representation.getBody()
                                        .filter(body -> body instanceof MarshaledResponseBody.Bytes || body instanceof MarshaledResponseBody.ByteBuffer)
                                        .orElse(null);
                        Long bodyLength = representation.getBody().map(MarshaledResponseBody::getLength).orElse(null);
                        requestResultHolder.updateAndGet(result -> result == null
                                        ? HttpRequestResult.withMarshaledResponse(representation).resourceMethod(resourceMethodHolder.get())
                                                        .headResponseCompressionBody(compressionBody).headResponseBodyLength(bodyLength).build()
                                        : result.copy().marshaledResponse(representation).headResponseCompressionBody(compressionBody)
                                                        .headResponseBodyLength(bodyLength).finish());
                };

                // Holders to permit mutable effectively-final state tracking
                AtomicBoolean willStartResponseWritingCompleted = new AtomicBoolean(false);
                AtomicBoolean didFinishResponseWritingCompleted = new AtomicBoolean(false);
                AtomicBoolean didFinishRequestHandlingCompleted = new AtomicBoolean(false);
                AtomicBoolean didInvokeWrapRequestConsumer = new AtomicBoolean(false);
                AtomicBoolean wrapRequestScopeOpen = new AtomicBoolean(true);
                Thread wrapRequestThread = Thread.currentThread();

                List<Throwable> throwables = new ArrayList<>(10);

                // A built-in transport can replace the logical response before commitment.
                // Keep write/finish observation aligned with that finite wire response.
                HttpResponseStreamObservation streamObservation = new HttpResponseStreamObservation(request, processingStartedNanos);
                HttpTransportResponseFeedback feedback = HttpTransportResponseFeedback.forConsumer(requestResultConsumer);
                if (feedback != null)
                        feedback.claimLifecycleConsumer();
                Consumer<HttpRequestResult> transportResultConsumer = requestResult -> {
                        if (serverType == ServerType.SSE)
                                requestResult = requestResult.copy().effectiveSseRequest(requestHolder.get()).finish();
                        if (serverType == ServerType.HTTP)
                                requestResult = requestResult.copy().responseStreamObservation(streamObservation).finish();
                        Throwable requestHandlingFailure = requestHandlingFailureHolder.get();
                        if (serverType == ServerType.SSE && requestHandlingFailure != null)
                                requestResult = requestResult.copy().sseHandshakeResult(null)
                                                .requestHandlingFailure(requestHandlingFailure).finish();
                        else if (serverType == ServerType.SSE
                                        && requestResult.getSseHandshakeResult().orElse(null) instanceof SseHandshakeResult.Accepted
                                        && !isCompatibleAcceptedSseResponse(requestResult.getMarshaledResponse()))
                                // An interceptor can replace an accepted handshake with a finite denial.
                                // The final response owns that decision in both live and simulated transports.
                                requestResult = requestResult.copy().sseHandshakeResult(null).finish();
                        HttpTransportResponseOutcome compatibilityOutcome = HttpTransportResponseOutcome.accepted();
                        try {
                                requestResultConsumer.accept(requestResult);
                        } catch (HttpTransportResponseFeedback.UnclaimedReplacement replacement) {
                                compatibilityOutcome = replacement.getOutcome();
                        } finally {
                                // Consume feedback even when a decorating callback fails after delivery.
                                HttpTransportResponseOutcome outcome = feedback == null ? compatibilityOutcome : feedback.finish();
                                if (outcome instanceof HttpTransportResponseOutcome.Accepted)
                                        outcome = compatibilityOutcome;
                                switch (outcome) {
                                        case HttpTransportResponseOutcome.Accepted ignored -> {}
                                        case HttpTransportResponseOutcome.NotWritten notWritten -> notWritten.rethrowWriteFailure();
                                        case HttpTransportResponseOutcome.Replaced replacement -> {
                                                marshaledResponseHolder.set(replacement.getMarshaledResponse());
                                                requestResultHolder.set(requestResult.copy().marshaledResponse(replacement.getMarshaledResponse())
                                                                .headResponseCompressionBody(null).headResponseBodyLength(null).finish());
                                                replacement.getPreparationFailure().ifPresent(throwables::add);
                                                replacement.rethrowWriteFailure();
                                        }
                                }
                        }
                };

                Consumer<LogEvent> safelyLog = (logEvent -> {
                        try {
                                lifecycleObserver.didReceiveLogEvent(logEvent);
                        } catch (Throwable throwable) {
                                LifecycleObserverLogFallback.report(throwable);
                                throwables.add(throwable);
                        }
                });

                BiConsumer<String, Consumer<MetricsCollector>> safelyCollectMetrics = (message, metricsInvocation) -> {
                        if (metricsCollector == null)
                                return;

                        try {
                                metricsInvocation.accept(metricsCollector);
                        } catch (Throwable throwable) {
                                safelyLog.accept(LogEvent.with(LogEventType.METRICS_COLLECTOR_FAILED, message)
                                                .throwable(throwable)
                                                .request(requestHolder.get())
                                                .resourceMethod(resourceMethodHolder.get())
                                                .marshaledResponse(marshaledResponseHolder.get())
                                                .build());
                        }
                };

                AtomicBoolean didStartRequestHandling = new AtomicBoolean();
                Runnable startRequestHandling = () -> {
                        if (!didStartRequestHandling.compareAndSet(false, true))
                                return;

                        try {
                                lifecycleObserver.didStartRequestHandling(serverType, request, resourceMethodHolder.get());
                        } catch (Throwable t) {
                                safelyLog.accept(LogEvent.with(LogEventType.LIFECYCLE_OBSERVER_DID_START_REQUEST_HANDLING_FAILED,
                                                                format("An exception occurred while invoking %s::didStartRequestHandling",
                                                                                LifecycleObserver.class.getSimpleName()))
                                                .throwable(t)
                                                .request(requestHolder.get())
                                                .resourceMethod(resourceMethodHolder.get())
                                                .build());

                                throwables.add(t);
                        }

                        safelyCollectMetrics.accept(
                                        format("An exception occurred while invoking %s::didStartRequestHandling", MetricsCollector.class.getSimpleName()),
                                        (metricsInvocation) -> metricsInvocation.didStartRequestHandling(serverType, request, resourceMethodHolder.get()));
                };

                requestHolder.set(request);

                try {
                        try {
                        requestInterceptor.wrapRequest(serverType, request, (wrappedRequest) -> {
                                if (!wrapRequestScopeOpen.get() || !sameInstance(Thread.currentThread(), wrapRequestThread))
                                        throw new IllegalStateException("RequestInterceptor::wrapRequest must call requestProcessor synchronously before returning");
                                if (!didInvokeWrapRequestConsumer.compareAndSet(false, true))
                                        throw new IllegalStateException("RequestInterceptor::wrapRequest must call requestProcessor exactly once");
                                requestHolder.set(wrappedRequest);

                                try {
                                        // Resolve after wrapping so path/method rewrites affect routing.
                                        ResourceMethod resolvedResourceMethod =
                                                        resolveResourceMethodForRequest(requireNonNull(requestHolder.get()), resourceMethodResolver, serverType);
                                        resourceMethodHolder.set(resolvedResourceMethod);
                                        resourceMethodResolutionExceptionHolder.set(null);
                                } catch (Throwable t) {
                                        safelyLog.accept(LogEvent.with(LogEventType.RESOURCE_METHOD_RESOLUTION_FAILED, "Unable to resolve Resource Method")
                                                        .throwable(t)
                                                        .request(requestHolder.get())
                                                        .build());

                                        // If an exception occurs here, keep track of it - we will surface them after letting LifecycleObserver
                                        // see that a request has come in.
                                        throwables.add(t);
                                        resourceMethodResolutionExceptionHolder.set(t);
                                        resourceMethodHolder.set(null);
                                }

                                startRequestHandling.run();

                                // Retain generated file bodies until the interceptor chooses its final response.
                                // The list allocates storage only for responses that actually own a channel.
                                List<MarshaledResponse> pendingResponseBodies = new ArrayList<>(0);
                                try {
                                        AtomicBoolean didInvokeMarshaledResponseConsumer = new AtomicBoolean(false);

                                        requestInterceptor.interceptRequest(serverType, requestHolder.get(), resourceMethodHolder.get(), (interceptorRequest) -> {
                                                requestHolder.set(interceptorRequest);

                                                try {
                                                        if (resourceMethodResolutionExceptionHolder.get() != null)
                                                                throw resourceMethodResolutionExceptionHolder.get();

                                                        HttpRequestResult requestResult = toHttpRequestResult(requestHolder.get(), resourceMethodHolder.get(), serverType);
                                                        requestResultHolder.set(requestResult);

                                                        // The interceptor chooses the response before CORS, postprocessing and HEAD projection.
                                                        MarshaledResponse updatedMarshaledResponse = requestResult.getMarshaledResponse();
                                                        return ResponseBodyCleanup.track(pendingResponseBodies, updatedMarshaledResponse);
                                                        } catch (Throwable t) {
                                                                requestHandlingFailureHolder.compareAndSet(null, t);
                                                                requestResultHolder.updateAndGet(result -> result == null ? null
                                                                                : result.copy().headResponseCompressionBody(null).headResponseBodyLength(null).finish());
                                                                if (!sameInstance(t, resourceMethodResolutionExceptionHolder.get())) {
                                                                        throwables.add(t);

                                                                safelyLog.accept(LogEvent.with(LogEventType.REQUEST_PROCESSING_FAILED,
                                                                                                "An exception occurred while processing request")
                                                                                .throwable(t)
                                                                                .request(requestHolder.get())
                                                                                .resourceMethod(resourceMethodHolder.get())
                                                                                .build());
                                                        }

                                                        // Unhappy path.  Try to use configuration's exception response marshaler...
                                                        try {
                                                                MarshaledResponse marshaledResponse = responseMarshaler.forThrowable(requestHolder.get(), t, resourceMethodHolder.get());
                                                                marshaledResponseHolder.set(marshaledResponse);

                                                                return ResponseBodyCleanup.track(pendingResponseBodies, marshaledResponse);
                                                        } catch (Throwable t2) {
                                                                throwables.add(t2);

                                                                safelyLog.accept(LogEvent.with(LogEventType.RESPONSE_MARSHALER_FOR_THROWABLE_FAILED,
                                                                                                "An exception occurred while trying to write an exception response")
                                                                                .throwable(t2)
                                                                                .request(requestHolder.get())
                                                                                .resourceMethod(resourceMethodHolder.get())
                                                                                .build());

                                                                // The configuration's exception response marshaler failed - provide a failsafe response to recover
                                                                return ResponseBodyCleanup.track(pendingResponseBodies, provideFailsafeMarshaledResponse(requestHolder.get(), t2));
                                                        }
                                                }
                                        }, (interceptorMarshaledResponse) -> {
                                                requireNonNull(interceptorMarshaledResponse);
                                                ResponseBodyCleanup.track(pendingResponseBodies, interceptorMarshaledResponse);
                                                enforceMarshaledResponseStatusCode(interceptorMarshaledResponse, false);
                                                didInvokeMarshaledResponseConsumer.set(true);
                                                HttpRequestResult requestResult = requestResultHolder.get();
                                                if (serverType == ServerType.SSE && requestResult != null
                                                                && requestResult.getSseHandshakeResult().orElse(null) instanceof SseHandshakeResult.Accepted
                                                                && !isCompatibleAcceptedSseResponse(interceptorMarshaledResponse)) {
                                                        requestResult = requestResult.copy().sseHandshakeResult(null).finish();
                                                        requestResultHolder.set(requestResult);
                                                }
                                                if (requestResult != null && !requestResult.getMarshaledResponse().getStatusCode()
                                                                .equals(interceptorMarshaledResponse.getStatusCode()))
                                                        requestResultHolder.set(requestResult.copy().headResponseCompressionBody(null).headResponseBodyLength(null).finish());
                                                marshaledResponseHolder.set(interceptorMarshaledResponse);
                                        });

                                        if (!didInvokeMarshaledResponseConsumer.get()) {
                                                requestResultHolder.set(null);
                                                throw new IllegalStateException(format("%s::interceptRequest must call responseWriter", RequestInterceptor.class.getSimpleName()));
                                        }
                                } catch (Throwable t) {
                                        requestHandlingFailureHolder.compareAndSet(null, t);
                                        throwables.add(t);
                                        requestResultHolder.updateAndGet(result -> result == null ? null
                                                        : result.copy().headResponseCompressionBody(null).headResponseBodyLength(null).finish());

                                        try {
                                                // In the event that an error occurs during processing of a RequestInterceptor method, for example
                                                safelyLog.accept(LogEvent.with(LogEventType.REQUEST_INTERCEPTOR_INTERCEPT_REQUEST_FAILED,
                                                                                format("An exception occurred while invoking %s::interceptRequest", RequestInterceptor.class.getSimpleName()))
                                                                .throwable(t)
                                                                .request(requestHolder.get())
                                                                .resourceMethod(resourceMethodHolder.get())
                                                                .build());

                                                MarshaledResponse marshaledResponse = responseMarshaler.forThrowable(requestHolder.get(), t, resourceMethodHolder.get());
                                                marshaledResponseHolder.set(marshaledResponse);
                                        } catch (Throwable t2) {
                                                throwables.add(t2);

                                                safelyLog.accept(LogEvent.with(LogEventType.RESPONSE_MARSHALER_FOR_THROWABLE_FAILED,
                                                                                format("An exception occurred while invoking %s::forThrowable when trying to write an exception response", ResponseMarshaler.class.getSimpleName()))
                                                                .throwable(t2)
                                                                .request(requestHolder.get())
                                                                .resourceMethod(resourceMethodHolder.get())
                                                                .build());

                                                marshaledResponseHolder.set(provideFailsafeMarshaledResponse(requestHolder.get(), t2));
                                        }
                                } finally {
                                        // A broken exception marshaler must not turn missing response selection
                                        // into a transport write failure or expose null responses to observers.
                                        if (marshaledResponseHolder.get() == null) {
                                                Throwable failure = requestHandlingFailureHolder.get();
                                                if (failure == null)
                                                        failure = new IllegalStateException("Request handling did not select a response");
                                                marshaledResponseHolder.set(provideFailsafeMarshaledResponse(
                                                                requestHolder.get() == null ? request : requestHolder.get(), failure));
                                        }
                                        MarshaledResponse selectedResponse = marshaledResponseHolder.get();
                                        if (selectedResponse != null && requiresPostProcessing(selectedResponse, postProcessingToken)) {
                                                try {
                                                        HttpRequestResult selectedResult = requestResultHolder.get();
                                                        boolean acceptedSse = serverType == ServerType.SSE && selectedResult != null
                                                                        && selectedResult.getSseHandshakeResult().orElse(null) instanceof SseHandshakeResult.Accepted;
                                                        MarshaledResponse finalized = ResponseBodyCleanup.transform(selectedResponse,
                                                                        response -> applyCommonPropertiesToMarshaledResponse(requestHolder.get(), response, acceptedSse, captureHeadRepresentation, postProcessingToken));
                                                        try {
                                                                if (acceptedSse) isCompatibleAcceptedSseResponse(finalized);
                                                        } catch (RuntimeException | Error failure) {
                                                                ResponseBodyCleanup.discard(finalized, null, failure);
                                                                throw failure;
                                                        }
                                                        marshaledResponseHolder.set(finalized);
                                                } catch (Throwable failure) {
                                                        throwables.add(failure);
                                                        requestHandlingFailureHolder.compareAndSet(null, failure);
                                                        requestResultHolder.updateAndGet(result -> result == null ? null
                                                                        : result.copy().headResponseCompressionBody(null).headResponseBodyLength(null).finish());
                                                        safelyLog.accept(LogEvent.with(LogEventType.REQUEST_PROCESSING_FAILED,
                                                                        "An exception occurred while finalizing response").throwable(failure)
                                                                        .request(requestHolder.get()).resourceMethod(resourceMethodHolder.get()).build());
                                                        try {
                                                                MarshaledResponse recovery = responseMarshaler.forThrowable(requestHolder.get(), failure, resourceMethodHolder.get());
                                                                recovery = ResponseBodyCleanup.transform(recovery, response -> applyCommonPropertiesToMarshaledResponse(
                                                                                requestHolder.get(), response, false, ignored -> {}, postProcessingToken, false));
                                                                marshaledResponseHolder.set(recovery);
                                                        } catch (Throwable recoveryFailure) {
                                                                throwables.add(recoveryFailure);
                                                                safelyLog.accept(LogEvent.with(LogEventType.RESPONSE_MARSHALER_FOR_THROWABLE_FAILED,
                                                                                "Unable to finalize exception response").throwable(recoveryFailure)
                                                                                .request(requestHolder.get()).resourceMethod(resourceMethodHolder.get()).build());
                                                                marshaledResponseHolder.set(provideFailsafeMarshaledResponse(requestHolder.get(), recoveryFailure));
                                                        }
                                                }
                                        }
                                        for (MarshaledResponse pendingResponse : pendingResponseBodies) {
                                                try {
                                                        MarshaledResponse selected = marshaledResponseHolder.get();
                                                        if (selected == null) pendingResponse.close();
                                                        else marshaledResponseHolder.set(ResponseBodyCleanup.retain(pendingResponse, selected));
                                                } catch (Throwable cleanupFailure) {
                                                        throwables.add(cleanupFailure);
                                                        safelyLog.accept(LogEvent.with(LogEventType.REQUEST_PROCESSING_FAILED,
                                                                        "Unable to release an abandoned response body")
                                                                        .throwable(cleanupFailure).request(requestHolder.get())
                                                                        .resourceMethod(resourceMethodHolder.get()).build());
                                                }
                                        }
                                        pendingResponseBodies.clear();
                                        try {
                                                try {
                                                        lifecycleObserver.willWriteResponse(serverType, requestHolder.get(), resourceMethodHolder.get(), marshaledResponseHolder.get());
                                                } catch (Throwable t) {
                                                        throwables.add(t);
                                                        safelyLog.accept(LogEvent.with(LogEventType.LIFECYCLE_OBSERVER_WILL_WRITE_RESPONSE_FAILED,
                                                                        format("An exception occurred while invoking %s::willWriteResponse", LifecycleObserver.class.getSimpleName()))
                                                                        .throwable(t).request(requestHolder.get()).resourceMethod(resourceMethodHolder.get())
                                                                        .marshaledResponse(marshaledResponseHolder.get()).build());
                                                } finally {
                                                        willStartResponseWritingCompleted.set(true);
                                                }

                                                safelyCollectMetrics.accept(
                                                                format("An exception occurred while invoking %s::willWriteResponse", MetricsCollector.class.getSimpleName()),
                                                                (metricsInvocation) -> metricsInvocation.willWriteResponse(serverType, requestHolder.get(), resourceMethodHolder.get(), marshaledResponseHolder.get()));

                                                Instant responseWriteStarted = Instant.now();

                                                try {
                                                        HttpRequestResult requestResult = requestResultHolder.get();

                                                        if (requestResult != null)
                                                                transportResultConsumer.accept(requestResult.copy()
                                                                                .marshaledResponse(requireNonNull(marshaledResponseHolder.get())).finish());
                                                        else
                                                                transportResultConsumer.accept(HttpRequestResult.withMarshaledResponse(marshaledResponseHolder.get())
                                                                                .resourceMethod(resourceMethodHolder.get())
                                                                                .build());

                                                        Instant responseWriteFinished = Instant.now();
                                                        Duration responseWriteDuration = Duration.between(responseWriteStarted, responseWriteFinished);

                                                        try {
                                                                lifecycleObserver.didWriteResponse(serverType, requestHolder.get(), resourceMethodHolder.get(), marshaledResponseHolder.get(), responseWriteDuration);
                                                        } catch (Throwable t) {
                                                                throwables.add(t);

                                                                safelyLog.accept(LogEvent.with(LogEventType.LIFECYCLE_OBSERVER_DID_WRITE_RESPONSE_FAILED,
                                                                                                format("An exception occurred while invoking %s::didWriteResponse",
                                                                                                                LifecycleObserver.class.getSimpleName()))
                                                                                .throwable(t)
                                                                                .request(requestHolder.get())
                                                                                .resourceMethod(resourceMethodHolder.get())
                                                                                .marshaledResponse(marshaledResponseHolder.get())
                                                                                .build());
                                                        } finally {
                                                                didFinishResponseWritingCompleted.set(true);
                                                        }

                                                        safelyCollectMetrics.accept(
                                                                        format("An exception occurred while invoking %s::didWriteResponse", MetricsCollector.class.getSimpleName()),
                                                                        (metricsInvocation) -> metricsInvocation.didWriteResponse(serverType, requestHolder.get(), resourceMethodHolder.get(),
                                                                                        marshaledResponseHolder.get(), responseWriteDuration));
                                                } catch (Throwable t) {
                                                        didFinishResponseWritingCompleted.set(true);
                                                        throwables.add(t);

                                                        Instant responseWriteFinished = Instant.now();
                                                        Duration responseWriteDuration = Duration.between(responseWriteStarted, responseWriteFinished);

                                                        try {
                                                                lifecycleObserver.didFailToWriteResponse(serverType, requestHolder.get(), resourceMethodHolder.get(), marshaledResponseHolder.get(), responseWriteDuration, t);
                                                        } catch (Throwable t2) {
                                                                throwables.add(t2);

                                                                safelyLog.accept(LogEvent.with(LogEventType.LIFECYCLE_OBSERVER_DID_WRITE_RESPONSE_FAILED,
                                                                                                format("An exception occurred while invoking %s::didFailToWriteResponse",
                                                                                                                LifecycleObserver.class.getSimpleName()))
                                                                                .throwable(t2)
                                                                                .request(requestHolder.get())
                                                                                .resourceMethod(resourceMethodHolder.get())
                                                                                .marshaledResponse(marshaledResponseHolder.get())
                                                                                .build());
                                                        }

                                                        safelyCollectMetrics.accept(
                                                                        format("An exception occurred while invoking %s::didFailToWriteResponse", MetricsCollector.class.getSimpleName()),
                                                                        (metricsInvocation) -> metricsInvocation.didFailToWriteResponse(serverType, requestHolder.get(), resourceMethodHolder.get(),
                                                                                        marshaledResponseHolder.get(), responseWriteDuration, t));
                                                }
                                        } finally {
                                                Duration processingDuration = Duration.ofNanos(Math.max(0L,
                                                                System.nanoTime() - processingStartedNanos));

                                                safelyCollectMetrics.accept(
                                                                format("An exception occurred while invoking %s::didFinishRequestHandling", MetricsCollector.class.getSimpleName()),
                                                                (metricsInvocation) -> metricsInvocation.didFinishRequestHandling(serverType, request, resourceMethodHolder.get(), marshaledResponseHolder.get(), processingDuration, Collections.unmodifiableList(throwables)));

                                        MarshaledResponse finishedResponse = marshaledResponseHolder.get();
                                        streamObservation.completeHandling(serverType == ServerType.HTTP && finishedResponse != null && finishedResponse.isStreaming());

                                                try {
                                                        lifecycleObserver.didFinishRequestHandling(serverType, request, resourceMethodHolder.get(), marshaledResponseHolder.get(), processingDuration, Collections.unmodifiableList(throwables));
                                                } catch (Throwable t) {
                                                        safelyLog.accept(LogEvent.with(LogEventType.LIFECYCLE_OBSERVER_DID_FINISH_REQUEST_HANDLING_FAILED,
                                                                                        format("An exception occurred while invoking %s::didFinishRequestHandling",
                                                                                                        LifecycleObserver.class.getSimpleName()))
                                                                        .throwable(t)
                                                                        .request(requestHolder.get())
                                                                        .resourceMethod(resourceMethodHolder.get())
                                                                        .marshaledResponse(marshaledResponseHolder.get())
                                                                        .build());
                                                } finally {
                                                        didFinishRequestHandlingCompleted.set(true);
                                                }
                                        }
                                }
                        });
                        } finally {
                                wrapRequestScopeOpen.set(false);
                        }

                        if (!didInvokeWrapRequestConsumer.get())
                                throw new IllegalStateException(format("%s::wrapRequest must call requestProcessor", RequestInterceptor.class.getSimpleName()));
                } catch (Throwable t) {
                        requestHandlingFailureHolder.compareAndSet(null, t);
                        if (!didFinishRequestHandlingCompleted.get())
                                throwables.add(t);
                        // If an error occurred during request wrapping, it's possible a response was never written/communicated back to LifecycleObserver.
                        // Detect that here and inform LifecycleObserver accordingly.
                        safelyLog.accept(LogEvent.with(LogEventType.REQUEST_INTERCEPTOR_WRAP_REQUEST_FAILED,
                                                        format("An exception occurred while invoking %s::wrapRequest",
                                                                        RequestInterceptor.class.getSimpleName()))
                                        .throwable(t)
                                        .request(requestHolder.get())
                                        .resourceMethod(resourceMethodHolder.get())
                                        .marshaledResponse(marshaledResponseHolder.get())
                                        .build());

                        // Wrapping can fail before dispatch; keep start/finish callbacks paired for observers and metrics.
                        startRequestHandling.run();

                        requestResultHolder.updateAndGet(result -> result == null ? null
                                        : result.copy().headResponseCompressionBody(null).headResponseBodyLength(null).finish());

                        // If we don't have a response, let the marshaler try to make one for the exception.
                        // If that fails, use the failsafe.
                        if (marshaledResponseHolder.get() == null) {
                                try {
                                        MarshaledResponse marshaledResponse = responseMarshaler.forThrowable(requestHolder.get(), t, resourceMethodHolder.get());
                                        marshaledResponse = ResponseBodyCleanup.transform(marshaledResponse,
                                                        response -> applyCommonPropertiesToMarshaledResponse(requestHolder.get(), response, false, ignored -> {}, postProcessingToken, false));
                                        marshaledResponseHolder.set(marshaledResponse);
                                } catch (Throwable t2) {
                                        throwables.add(t2);

                                        safelyLog.accept(LogEvent.with(LogEventType.RESPONSE_MARSHALER_FOR_THROWABLE_FAILED,
                                                                        format("An exception occurred during request wrapping while invoking %s::forThrowable",
                                                                                        ResponseMarshaler.class.getSimpleName()))
                                                        .throwable(t2)
                                                        .request(requestHolder.get())
                                                        .resourceMethod(resourceMethodHolder.get())
                                                        .marshaledResponse(marshaledResponseHolder.get())
                                                        .build());

                                        marshaledResponseHolder.set(provideFailsafeMarshaledResponse(requestHolder.get(), t));
                                }
                        }

                        if (!willStartResponseWritingCompleted.get()) {
                                try {
                                        lifecycleObserver.willWriteResponse(serverType, requestHolder.get(), resourceMethodHolder.get(), marshaledResponseHolder.get());
                                } catch (Throwable t2) {
                                        safelyLog.accept(LogEvent.with(LogEventType.LIFECYCLE_OBSERVER_WILL_WRITE_RESPONSE_FAILED,
                                                                        format("An exception occurred while invoking %s::willWriteResponse",
                                                                                        LifecycleObserver.class.getSimpleName()))
                                                        .throwable(t2)
                                                        .request(requestHolder.get())
                                                        .resourceMethod(resourceMethodHolder.get())
                                                        .marshaledResponse(marshaledResponseHolder.get())
                                                        .build());
                                }

                                safelyCollectMetrics.accept(
                                                format("An exception occurred while invoking %s::willWriteResponse", MetricsCollector.class.getSimpleName()),
                                                (metricsInvocation) -> metricsInvocation.willWriteResponse(serverType, requestHolder.get(), resourceMethodHolder.get(), marshaledResponseHolder.get()));
                        }

                        try {
                                Instant responseWriteStarted = Instant.now();

                                if (!didFinishResponseWritingCompleted.get()) {
                                        try {
                                                HttpRequestResult requestResult = requestResultHolder.get();

                                                if (requestResult != null)
                                                        transportResultConsumer.accept(requestResult.copy()
                                                                        .marshaledResponse(requireNonNull(marshaledResponseHolder.get())).finish());
                                                else
                                                        transportResultConsumer.accept(HttpRequestResult.withMarshaledResponse(marshaledResponseHolder.get())
                                                                        .resourceMethod(resourceMethodHolder.get())
                                                                        .build());

                                                Instant responseWriteFinished = Instant.now();
                                                Duration responseWriteDuration = Duration.between(responseWriteStarted, responseWriteFinished);

                                                try {
                                                        lifecycleObserver.didWriteResponse(serverType, requestHolder.get(), resourceMethodHolder.get(), marshaledResponseHolder.get(), responseWriteDuration);
                                                } catch (Throwable t2) {
                                                        throwables.add(t2);

                                                        safelyLog.accept(LogEvent.with(LogEventType.LIFECYCLE_OBSERVER_DID_WRITE_RESPONSE_FAILED,
                                                                                        format("An exception occurred while invoking %s::didWriteResponse",
                                                                                                        LifecycleObserver.class.getSimpleName()))
                                                                        .throwable(t2)
                                                                        .request(requestHolder.get())
                                                                        .resourceMethod(resourceMethodHolder.get())
                                                                        .marshaledResponse(marshaledResponseHolder.get())
                                                                        .build());
                                                }

                                                safelyCollectMetrics.accept(
                                                                format("An exception occurred while invoking %s::didWriteResponse", MetricsCollector.class.getSimpleName()),
                                                                (metricsInvocation) -> metricsInvocation.didWriteResponse(serverType, requestHolder.get(), resourceMethodHolder.get(),
                                                                                marshaledResponseHolder.get(), responseWriteDuration));
                                        } catch (Throwable t2) {
                                                throwables.add(t2);

                                                Instant responseWriteFinished = Instant.now();
                                                Duration responseWriteDuration = Duration.between(responseWriteStarted, responseWriteFinished);

                                                try {
                                                        lifecycleObserver.didFailToWriteResponse(serverType, requestHolder.get(), resourceMethodHolder.get(), marshaledResponseHolder.get(), responseWriteDuration, t2);
                                                } catch (Throwable t3) {
                                                        throwables.add(t3);

                                                        safelyLog.accept(LogEvent.with(LogEventType.LIFECYCLE_OBSERVER_DID_WRITE_RESPONSE_FAILED,
                                                                                        format("An exception occurred while invoking %s::didFailToWriteResponse",
                                                                                                        LifecycleObserver.class.getSimpleName()))
                                                                        .throwable(t3)
                                                                        .request(requestHolder.get())
                                                                        .resourceMethod(resourceMethodHolder.get())
                                                                        .marshaledResponse(marshaledResponseHolder.get())
                                                                        .build());
                                                }

                                                safelyCollectMetrics.accept(
                                                                format("An exception occurred while invoking %s::didFailToWriteResponse", MetricsCollector.class.getSimpleName()),
                                                                (metricsInvocation) -> metricsInvocation.didFailToWriteResponse(serverType, requestHolder.get(), resourceMethodHolder.get(),
                                                                                marshaledResponseHolder.get(), responseWriteDuration, t2));
                                        }
                                }
                        } finally {
                                if (!didFinishRequestHandlingCompleted.get()) {
                                        Duration processingDuration = Duration.ofNanos(Math.max(0L,
                                                        System.nanoTime() - processingStartedNanos));

                                        safelyCollectMetrics.accept(
                                                        format("An exception occurred while invoking %s::didFinishRequestHandling", MetricsCollector.class.getSimpleName()),
                                                        (metricsInvocation) -> metricsInvocation.didFinishRequestHandling(serverType, request, resourceMethodHolder.get(), marshaledResponseHolder.get(), processingDuration, Collections.unmodifiableList(throwables)));

                                        MarshaledResponse finishedResponse = marshaledResponseHolder.get();
                                        streamObservation.completeHandling(serverType == ServerType.HTTP && finishedResponse != null && finishedResponse.isStreaming());

                                        try {
                                                lifecycleObserver.didFinishRequestHandling(serverType, request, resourceMethodHolder.get(), marshaledResponseHolder.get(), processingDuration, Collections.unmodifiableList(throwables));
                                        } catch (Throwable t2) {
                                                safelyLog.accept(LogEvent.with(LogEventType.LIFECYCLE_OBSERVER_DID_FINISH_REQUEST_HANDLING_FAILED,
                                                                                format("An exception occurred while invoking %s::didFinishRequestHandling",
                                                                                                LifecycleObserver.class.getSimpleName()))
                                                                .throwable(t2)
                                                                .request(requestHolder.get())
                                                                .resourceMethod(resourceMethodHolder.get())
                                                                .marshaledResponse(marshaledResponseHolder.get())
                                                                .build());
                                        }
                                }
                        }
                }
        }

        @NonNull
        HttpRequestResult toHttpRequestResult(@NonNull Request request,
                                                                                                                                                                        @Nullable ResourceMethod resourceMethod,
                                                                                                                                                                        @NonNull ServerType serverType) throws Throwable {
                requireNonNull(request);
                requireNonNull(serverType);

                ResourceMethodParameterProvider resourceMethodParameterProvider = getSokletConfig().getResourceMethodParameterProvider();
                InstanceProvider instanceProvider = getSokletConfig().getInstanceProvider();
                CorsAuthorizer corsAuthorizer = getSokletConfig().getCorsAuthorizer();
                ResourceMethodResolver resourceMethodResolver = getSokletConfig().getResourceMethodResolver();
                ResponseMarshaler responseMarshaler = getSokletConfig().getResponseMarshaler();
                CorsPreflight corsPreflight = request.getCorsPreflight().orElse(null);

                // Special short-circuit for big requests
                if (request.isContentTooLarge()) {
                        ResourceMethod contentTooLargeResourceMethod =
                                        resolveResourceMethodForRequest(request, resourceMethodResolver, serverType);
                        return HttpRequestResult.withMarshaledResponse(responseMarshaler.forContentTooLarge(request, contentTooLargeResourceMethod))
                                        .resourceMethod(resourceMethod)
                                        .build();
                }

                // Special short-circuit for OPTIONS *
                if (isOptionsSplat(request.getResourcePath()))
                        return HttpRequestResult.withMarshaledResponse(responseMarshaler.forOptionsSplat(request)).build();

                // CORS preflight authorization precedes explicit OPTIONS handlers too.
                if (resourceMethod == null || corsPreflight != null) {
                        // If this was an OPTIONS request, do special processing.
                        // If not, figure out if we should return a 404 or 405.
                        if (request.getHttpMethod() == HttpMethod.OPTIONS) {
                                // See what methods are available to us for this request's path
                                Map<HttpMethod, ResourceMethod> matchingResourceMethodsByHttpMethod = resolveMatchingResourceMethodsByHttpMethod(request, resourceMethodResolver, serverType);

                                // Special handling for CORS preflight requests, if needed
                                if (corsPreflight != null) {
                                        // Let configuration function determine if we should authorize this request.
                                        // Discard any OPTIONS references - see https://stackoverflow.com/a/68529748
                                        Map<HttpMethod, ResourceMethod> nonOptionsMatchingResourceMethodsByHttpMethod = matchingResourceMethodsByHttpMethod.entrySet().stream()
                                                        .filter(entry -> entry.getKey() != HttpMethod.OPTIONS)
                                                        .collect(Collectors.toMap(Entry::getKey, Entry::getValue));

                                        // Validate client-supplied reflection candidates before application
                                        // authorization, including custom authorizers that reflect them verbatim.
                                        CorsPreflightResponse corsPreflightResponse = corsPreflight.isSafeToReflect()
                                                        ? corsAuthorizer.authorizePreflight(request, corsPreflight, nonOptionsMatchingResourceMethodsByHttpMethod).orElse(null)
                                                        : null;

                                        // Allow or reject CORS depending on what the function said to do
                                        if (corsPreflightResponse != null) {
                                                // Allow
                                                MarshaledResponse marshaledResponse = responseMarshaler.forCorsPreflightAllowed(request, corsPreflight, corsPreflightResponse);

                                                return HttpRequestResult.withMarshaledResponse(marshaledResponse)
                                                                .corsPreflightResponse(corsPreflightResponse)
                                                                .build();
                                        }

                                        // Reject
                                        return HttpRequestResult.withMarshaledResponse(responseMarshaler.forCorsPreflightRejected(request, corsPreflight))
                                                        .build();
                                } else {
                                        // Just a normal OPTIONS response (non-CORS-preflight).
                                        if (matchingResourceMethodsByHttpMethod.isEmpty())
                                                return HttpRequestResult.withMarshaledResponse(responseMarshaler.forNotFound(request)).build();

                                        // If there's a matching OPTIONS resource method for this OPTIONS request, then invoke it.
                                        ResourceMethod optionsResourceMethod = matchingResourceMethodsByHttpMethod.get(HttpMethod.OPTIONS);

                                        if (optionsResourceMethod != null) {
                                                resourceMethod = optionsResourceMethod;
                                        } else {
                                                Set<HttpMethod> allowedHttpMethods = allowedHttpMethodsForResponse(matchingResourceMethodsByHttpMethod, true, serverType);

                                                return HttpRequestResult.withMarshaledResponse(responseMarshaler.forOptions(request, allowedHttpMethods))
                                                                .build();
                                        }
                                }
                        } else if (request.getHttpMethod() == HttpMethod.HEAD && serverType == ServerType.HTTP) {
                                // If there's a matching GET resource method for this HEAD request, then invoke it
                                Request headGetRequest = request.copy().httpMethod(HttpMethod.GET).finish();
                                ResourceMethod headGetResourceMethod =
                                                validateResolvedResourceMethod(resourceMethodResolver
                                                                .resourceMethodForRequest(headGetRequest, serverType)
                                                                .orElse(null));

                                if (headGetResourceMethod != null)
                                        resourceMethod = headGetResourceMethod;
                                else {
                                        Map<HttpMethod, ResourceMethod> otherMethods = resolveMatchingResourceMethodsByHttpMethod(
                                                        request, resourceMethodResolver, serverType);
                                        boolean resourceExists = otherMethods.keySet().stream().anyMatch(method -> method != HttpMethod.OPTIONS);
                                        return HttpRequestResult.withMarshaledResponse(resourceExists
                                                        ? responseMarshaler.forMethodNotAllowed(request, allowedHttpMethodsForResponse(otherMethods, true, serverType))
                                                        : responseMarshaler.forNotFound(request)).build();
                                }
                        } else {
                                // Not an OPTIONS request, so it's possible we have a 405. See if other HTTP methods match...
                                Map<HttpMethod, ResourceMethod> otherMatchingResourceMethodsByHttpMethod = resolveMatchingResourceMethodsByHttpMethod(request, resourceMethodResolver, serverType);

                                Set<HttpMethod> matchingNonOptionsHttpMethods = otherMatchingResourceMethodsByHttpMethod.keySet().stream()
                                                .filter(httpMethod -> httpMethod != HttpMethod.OPTIONS)
                                                .collect(Collectors.toSet());

                                if (matchingNonOptionsHttpMethods.size() > 0) {
                                        // ...if some do, it's a 405
                                        Set<HttpMethod> allowedHttpMethods = allowedHttpMethodsForResponse(otherMatchingResourceMethodsByHttpMethod, true, serverType);
                                        return HttpRequestResult.withMarshaledResponse(responseMarshaler.forMethodNotAllowed(request, allowedHttpMethods))
                                                        .build();
                                } else {
                                        // no matching resource method found, it's a 404
                                        return HttpRequestResult.withMarshaledResponse(responseMarshaler.forNotFound(request))
                                                        .build();
                                }
                        }
                }

                // Found a resource method - happy path.
                // 1. Get an instance of the resource class
                // 2. Get values to pass to the resource method on the resource class
                // 3. Invoke the resource method and use its return value to drive a response
                SokletFrameworkSetup.validateConfiguredResourceMethod(getSokletConfig(), resourceMethod);
                Class<?> resourceClass = resourceMethod.getMethod().getDeclaringClass();
                Object resourceClassInstance;

                try {
                        resourceClassInstance = instanceProvider.provide(resourceClass);
                } catch (Exception e) {
                        throw new IllegalArgumentException(format("Unable to acquire an instance of %s", resourceClass.getName()), e);
                }

                List<Object> parameterValues = resourceMethodParameterProvider.parameterValuesForResourceMethod(request, resourceMethod);

                Object responseObject;

                try {
                        responseObject = resourceMethod.getMethod().invoke(resourceClassInstance, parameterValues.toArray());
                } catch (InvocationTargetException e) {
                        if (e.getTargetException() != null)
                                throw e.getTargetException();

                        throw e;
                }

                // Unwrap the Optional<T>, if one exists.  We do not recurse deeper than one level
                if (responseObject instanceof Optional<?>)
                        responseObject = ((Optional<?>) responseObject).orElse(null);

                Response response;
                SseHandshakeResult sseHandshakeResult = null;

                // If null/void return, it's a 204
                // If it's a MarshaledResponse object, no marshaling + return it immediately - caller knows exactly what it wants to write.
                // If it's a Response object, use as is.
                // If it's a non-Response type of object, assume it's the response body and wrap in a Response.
                if (responseObject == null) {
                        response = Response.withStatusCode(204).build();
                } else if (responseObject instanceof MarshaledResponse) {
                        MarshaledResponse marshaledResponse = normalizeEmptyResetContent((MarshaledResponse) responseObject);
                        enforceMarshaledResponseStatusCode(marshaledResponse, true);

                        return HttpRequestResult.withMarshaledResponse(marshaledResponse)
                                        .resourceMethod(resourceMethod)
                                        .build();
                } else if (responseObject instanceof Response) {
                        response = (Response) responseObject;
                } else if (responseObject instanceof SseHandshakeResult.Accepted accepted) { // SSE "accepted" handshake
                        return HttpRequestResult.withMarshaledResponse(toMarshaledResponse(accepted))
                                        .resourceMethod(resourceMethod)
                                        .sseHandshakeResult(accepted)
                                        .build();
                } else if (responseObject instanceof SseHandshakeResult.Rejected rejected) { // SSE "rejected" handshake
                        response = rejected.getResponse();
                        sseHandshakeResult = rejected;
                } else {
                        response = Response.withStatusCode(200).body(responseObject).build();
                }

                if (response.getStatusCode() == 205) {
                        Object body = response.getBody().orElse(null);
                        if ((body instanceof String text && text.isEmpty()) || (body instanceof byte[] bytes && bytes.length == 0))
                                response = response.copy().body(null).finish();
                }
                enforceBodylessStatusCode(response.getStatusCode(), response.getBody().isPresent());

                MarshaledResponse marshaledResponse = normalizeEmptyResetContent(responseMarshaler.forResourceMethod(request, response, resourceMethod));

                enforceMarshaledResponseStatusCode(marshaledResponse, true);

                return HttpRequestResult.withMarshaledResponse(marshaledResponse)
                                .response(response)
                                .resourceMethod(resourceMethod)
                                .sseHandshakeResult(sseHandshakeResult)
                                .build();
        }

        @NonNull
        private MarshaledResponse toMarshaledResponse(SseHandshakeResult.@NonNull Accepted accepted) {
                requireNonNull(accepted);

                Map<String, List<String>> headers = accepted.getHeaders();
                Set<String> forbiddenHeaderNames = Set.of("content-length", "proxy-connection", "te", "trailer", "transfer-encoding", "upgrade");
                for (String headerName : headers.keySet())
                        if (forbiddenHeaderNames.contains(headerName.toLowerCase(java.util.Locale.ENGLISH)))
                                throw new IllegalArgumentException("Unsupported framing header for an accepted SSE handshake");
                LinkedCaseInsensitiveMap<List<String>> finalHeaders = new LinkedCaseInsensitiveMap<>(DEFAULT_ACCEPTED_HANDSHAKE_HEADERS.size() + headers.size());

                // Start with defaults
                for (Map.Entry<String, List<String>> e : DEFAULT_ACCEPTED_HANDSHAKE_HEADERS.entrySet())
                        finalHeaders.put(e.getKey(), e.getValue()); // values already unmodifiable

                // Overlay user-supplied headers (prefer user values on key collision)
                for (Map.Entry<String, List<String>> e : headers.entrySet()) {
                        String headerName = requireNonNull(e.getKey());
                        if (headerName.equalsIgnoreCase("Connection")
                                        || headerName.equalsIgnoreCase("Keep-Alive"))
                                continue;
                        // Defensively copy so callers can't mutate after construction
                        List<String> values = e.getValue() == null ? List.of() : List.copyOf(e.getValue());
                        finalHeaders.put(headerName, values);
                }
                List<String> contentTypes = finalHeaders.getOrDefault("Content-Type", List.of());
                if (contentTypes == null || contentTypes.size() != 1
                                || !contentTypes.get(0).split(";", 2)[0].trim().equalsIgnoreCase("text/event-stream"))
                        throw new IllegalArgumentException("An accepted SSE handshake requires exactly one text/event-stream Content-Type");

                return MarshaledResponse.withStatusCode(200)
                                .headers(finalHeaders)
                                .cookies(accepted.getCookies())
                                .build();
        }

        private static boolean isCompatibleAcceptedSseResponse(@NonNull MarshaledResponse response) {
                if (response.getStatusCode() != 200 || response.isStreaming() || response.getBody().isPresent())
                        return false;
                List<String> contentTypes = response.getHeaders().getOrDefault("Content-Type", List.of());
                boolean compatible = contentTypes.size() == 1
                                && contentTypes.get(0).split(";", 2)[0].trim().equalsIgnoreCase("text/event-stream");
                if (compatible) {
                        Set<String> forbidden = Set.of("content-length", "proxy-connection", "te", "trailer", "transfer-encoding", "upgrade");
                        for (String headerName : response.getHeaders().keySet())
                                if (forbidden.contains(headerName.toLowerCase(java.util.Locale.ENGLISH)))
                                        throw new IllegalArgumentException("Unsupported framing header for an accepted SSE handshake");
                }
                return compatible;
        }

        @NonNull
        private static MarshaledResponse normalizeEmptyResetContent(@NonNull MarshaledResponse response) {
                if (response.getStatusCode() == 205 && !response.isStreaming() && response.getBody().isPresent()
                                && response.getBodyLength() == 0L)
                        return response.copy().body((MarshaledResponseBody) null).finish();
                return response;
        }

        private static void enforceBodylessStatusCode(@NonNull Integer statusCode,
                                                                                                                                                                                                @NonNull Boolean hasBody) {
                requireNonNull(statusCode);
                requireNonNull(hasBody);
                enforceFinalResponseStatusCode(statusCode);

                if (hasBody && isBodylessStatusCode(statusCode))
                        throw new IllegalStateException(format("HTTP status code %d must not include a response body", statusCode));
        }

        private static void enforceMarshaledResponseStatusCode(@NonNull MarshaledResponse response,
                        boolean resourceMethodResponse) {
                try {
                        if (resourceMethodResponse)
                                enforceBodylessStatusCode(response.getStatusCode(),
                                                response.getBody().isPresent() || response.getStreamingResponseBody().isPresent());
                        else
                                enforceFinalResponseStatusCode(response.getStatusCode());
                } catch (RuntimeException | Error failure) {
                        // This response will never reach transport cleanup. Release only an
                        // explicitly owned finite body and keep validation as the primary failure.
                        try {
                                response.close();
                        } catch (RuntimeException | Error cleanupFailure) {
                                if (!sameInstance(cleanupFailure, failure))
                                        failure.addSuppressed(cleanupFailure);
                        }
                        throw failure;
                }
        }

        private static void enforceFinalResponseStatusCode(@NonNull Integer statusCode) {
                requireNonNull(statusCode);
                if (statusCode < 200 || statusCode > 599)
                        throw new IllegalStateException("Final HTTP response status code must be between 200 and 599.");
        }

        private static boolean isBodylessStatusCode(@NonNull Integer statusCode) {
                requireNonNull(statusCode);
                return (statusCode >= 100 && statusCode < 200) || statusCode == 204 || statusCode == 205 || statusCode == 304;
        }

        @NonNull
        MarshaledResponse applyHeadResponseIfApplicable(@NonNull Request request,
                                                                                                                                                                                                                                                @NonNull MarshaledResponse marshaledResponse) {
                if (request.getHttpMethod() != HttpMethod.HEAD)
                        return marshaledResponse;

                return getSokletConfig().getResponseMarshaler().forHead(request, marshaledResponse);
        }

        // Hat tip to Aslan Parçası and GrayStar
        @NonNull
        MarshaledResponse applyCommonPropertiesToMarshaledResponse(@NonNull Request request,
                                                                                                                                                                                                                                                                                         @NonNull MarshaledResponse marshaledResponse) {
                requireNonNull(request);
                requireNonNull(marshaledResponse);

                return applyCommonPropertiesToMarshaledResponse(request, marshaledResponse, false);
        }

        @NonNull
        MarshaledResponse applyCommonPropertiesToMarshaledResponse(@NonNull Request request,
                                                                                                                                                                                                                                                                                         @NonNull MarshaledResponse marshaledResponse,
                                                                                                                                                                                                                                                                                         @NonNull Boolean suppressContentLength) {
                return applyCommonPropertiesToMarshaledResponse(request, marshaledResponse, suppressContentLength, ignored -> {}, newPostProcessingToken());
        }

        @NonNull
        private static final class ResponseFinalizationToken {
                private boolean postProcessingStarted;
        }

        private @NonNull ResponseFinalizationToken newPostProcessingToken() {
                // Recovery must not repeat a processor that completed or failed before a later hook failed.
                return new ResponseFinalizationToken();
        }

        private boolean requiresPostProcessing(@NonNull MarshaledResponse response, @Nullable ResponseFinalizationToken token) {
                return token != null && !response.isPostProcessedFor(token);
        }

        @NonNull
        private MarshaledResponse applyCommonPropertiesToMarshaledResponse(@NonNull Request request,
                        @NonNull MarshaledResponse response, boolean suppressContentLength,
                        @NonNull Consumer<@NonNull MarshaledResponse> headRepresentationConsumer, @Nullable ResponseFinalizationToken postProcessingToken) {
                return applyCommonPropertiesToMarshaledResponse(request, response, suppressContentLength,
                                headRepresentationConsumer, postProcessingToken, true);
        }

        @NonNull
        private MarshaledResponse applyCommonPropertiesToMarshaledResponse(@NonNull Request request,
                        @NonNull MarshaledResponse response, boolean suppressContentLength,
                        @NonNull Consumer<@NonNull MarshaledResponse> headRepresentationConsumer, @Nullable ResponseFinalizationToken postProcessingToken, boolean applyHead) {
                requireNonNull(request);
                requireNonNull(response);
                boolean head = applyHead && request.getHttpMethod() == HttpMethod.HEAD;
                response = normalizeBodylessMarshaledResponse(request, response);
                response = ResponseBodyCleanup.transform(response, current -> applyCorsResponseIfApplicable(request, current));
                if (requiresPostProcessing(response, postProcessingToken)) {
                        ResponseFinalizationToken token = requireNonNull(postProcessingToken);
                        if (!token.postProcessingStarted) {
                                token.postProcessingStarted = true;
                                ResponseMarshaler marshaler = getSokletConfig().getResponseMarshaler();
                                response = ResponseBodyCleanup.transform(response, marshaler::postProcess);
                        }
                        response = response.markPostProcessed(token);
                }
                @Nullable ResponseFinalizationToken completedPostProcessingToken =
                                postProcessingToken != null && response.isPostProcessedFor(postProcessingToken)
                                                ? postProcessingToken : null;
                MarshaledResponse headRepresentation = response;
                if (head) {
                        response = ResponseBodyCleanup.transform(response, current -> applyHeadResponseIfApplicable(request, current));
                        // Only a projection of this representation may borrow its discarded body.
                        // A fresh descriptor or changed representation metadata belongs to the HEAD handler.
                        List<String> projectedLength = response.getHeaders().get("Content-Length");
                        boolean preservesRepresentation = completedPostProcessingToken != null
                                        && response.isPostProcessedFor(completedPostProcessingToken)
                                        && response.getStatusCode().equals(headRepresentation.getStatusCode())
                                        && Objects.equals(response.getHeaders().get("Content-Encoding"), headRepresentation.getHeaders().get("Content-Encoding"))
                                        && (projectedLength == null || headRepresentation.getBody().isEmpty()
                                                        || projectedLength.equals(List.of(Long.toString(headRepresentation.getBodyLength()))));
                        if (!preservesRepresentation || response.getBody().isPresent() || response.isStreaming())
                                headRepresentation = response;
                }
                headRepresentationConsumer.accept(headRepresentation);
                if (completedPostProcessingToken != null)
                        response = response.markPostProcessed(completedPostProcessingToken);
                response = normalizeBodylessMarshaledResponse(request, response);
                if (!suppressContentLength) response = applyContentLengthIfApplicable(request, response);
                if (!response.getHeaders().containsKey("Date"))
                        response = response.copy().headers(headers -> headers.put("Date", List.of(HttpDate.currentSecondHeaderValue()))).finish();
                return response;
        }

        @NonNull
        private static MarshaledResponse normalizeBodylessMarshaledResponse(@NonNull Request request,
                        @NonNull MarshaledResponse marshaledResponse) {
                enforceMarshaledResponseStatusCode(marshaledResponse, false);
                if (isBodylessStatusCode(marshaledResponse.getStatusCode())) {
                        int statusCode = marshaledResponse.getStatusCode();
                        marshaledResponse = marshaledResponse.copy().withoutBody().withoutStreamingResponseBody().headers(headers -> {
                                headers.remove("Transfer-Encoding");
                                if (statusCode == 205)
                                        headers.put("Content-Length", List.of("0"));
                                else if (statusCode == 204 || request.getHttpMethod() != HttpMethod.HEAD)
                                        headers.remove("Content-Length");
                        }).finish();
                }
                return marshaledResponse;
        }

        @NonNull
        MarshaledResponse applyContentLengthIfApplicable(@NonNull Request request,
                                                                                                                                                                                                                                                 @NonNull MarshaledResponse marshaledResponse) {
                requireNonNull(request);
                requireNonNull(marshaledResponse);

                if (marshaledResponse.isStreaming())
                        return marshaledResponse;

                Set<String> normalizedHeaderNames = marshaledResponse.getHeaders().keySet().stream()
                                .map(headerName -> headerName.toLowerCase(Locale.US))
                                .collect(Collectors.toSet());

                // If Content-Length is already specified, don't do anything
                if (normalizedHeaderNames.contains("content-length") || normalizedHeaderNames.contains("transfer-encoding"))
                        return marshaledResponse;

                if (shouldOmitAutomaticContentLength(request, marshaledResponse))
                        return marshaledResponse;

                // If Content-Length is not specified, specify as the number of bytes in the body
                return marshaledResponse.copy()
                                .headers((mutableHeaders) -> {
                                        String contentLengthHeaderValue = String.valueOf(marshaledResponse.getBodyLength());
                                        mutableHeaders.put("Content-Length", List.of(contentLengthHeaderValue));
                                }).finish();
        }

        private boolean shouldOmitAutomaticContentLength(@NonNull Request request,
                                                                                                                                                                                                         @NonNull MarshaledResponse marshaledResponse) {
                requireNonNull(request);
                requireNonNull(marshaledResponse);

                int statusCode = marshaledResponse.getStatusCode();

                if ((statusCode >= 100 && statusCode < 200) || statusCode == 204 || statusCode == 304)
                        return true;

                return statusCode != 205 && request.getHttpMethod() == HttpMethod.HEAD && marshaledResponse.getBodyLength() == 0L;
        }

        @NonNull
        MarshaledResponse applyCorsResponseIfApplicable(@NonNull Request request,
                                                                                                                                                                                                                                                @NonNull MarshaledResponse marshaledResponse) {
                requireNonNull(request);
                requireNonNull(marshaledResponse);

                CorsAuthorizer corsAuthorizer = getSokletConfig().getCorsAuthorizer();
                // Include every cache variant, even when this request omits Origin or is denied.
                if (!sameInstance(corsAuthorizer, CorsAuthorizer.rejectAllInstance()) && request.getCorsPreflight().isEmpty())
                        marshaledResponse = marshaledResponse.copy().headers(headers -> Utilities.addVaryHeader(headers, "Origin")).finish();

                Cors cors = request.getCors().orElse(null);

                // If non-CORS request, nothing further to do (note that CORS preflight was handled earlier)
                if (cors == null || !Cors.isSafeToReflectOrigin(cors.getOrigin()))
                        return marshaledResponse;

                // Does the authorizer say we are authorized?
                CorsResponse corsResponse = corsAuthorizer.authorize(request, cors).orElse(null);

                // Not authorized - don't apply CORS headers to the response
                if (corsResponse == null)
                        return marshaledResponse;

                // Authorized - OK, let's apply the headers to the response
                return getSokletConfig().getResponseMarshaler().forCorsAllowed(request, cors, corsResponse, marshaledResponse);
        }

        @NonNull
        Map<@NonNull HttpMethod, @NonNull ResourceMethod> resolveMatchingResourceMethodsByHttpMethod(@NonNull Request request,
                                                                                                                                                                                                                                                                                                                                                                                                                                 @NonNull ResourceMethodResolver resourceMethodResolver,
                                                                                                                                                                                                                                                                                                                                                                                                                                 @NonNull ServerType serverType) {
                requireNonNull(request);
                requireNonNull(resourceMethodResolver);
                requireNonNull(serverType);

                // Special handling for OPTIONS *
                if (isOptionsSplat(request.getResourcePath()))
                        return new LinkedHashMap<>();

                Map<HttpMethod, ResourceMethod> matchingResourceMethodsByHttpMethod = new LinkedHashMap<>(HttpMethod.values().length);

                for (HttpMethod httpMethod : HttpMethod.values()) {
                        // Make a quick copy of the request to see if other paths match
                        Request otherRequest = request.copy().httpMethod(httpMethod).finish();
                        ResourceMethod resourceMethod =
                                        validateResolvedResourceMethod(resourceMethodResolver
                                                        .resourceMethodForRequest(otherRequest, serverType)
                                                        .orElse(null));

                        if (resourceMethod != null)
                                matchingResourceMethodsByHttpMethod.put(httpMethod, resourceMethod);
                }

                return matchingResourceMethodsByHttpMethod;
        }

        @Nullable
        private static ResourceMethod resolveResourceMethodForRequest(@NonNull Request request,
                        @NonNull ResourceMethodResolver resourceMethodResolver, @NonNull ServerType serverType) {
                ResourceMethod resourceMethod = validateResolvedResourceMethod(resourceMethodResolver
                                .resourceMethodForRequest(request, serverType).orElse(null));
                if (resourceMethod == null && request.getHttpMethod() == HttpMethod.HEAD && serverType == ServerType.HTTP) {
                        // Resolve the GET fallback before observers/interceptors see routing metadata.
                        // Only the lookup copy becomes GET; the application still receives HEAD.
                        Request getRequest = request.copy().httpMethod(HttpMethod.GET).finish();
                        resourceMethod = validateResolvedResourceMethod(resourceMethodResolver
                                        .resourceMethodForRequest(getRequest, serverType).orElse(null));
                }
                return resourceMethod;
        }

        @Nullable
        private static ResourceMethod validateResolvedResourceMethod(
                        @Nullable ResourceMethod resourceMethod) {
                if (resourceMethod != null)
                        SokletFrameworkSetup.validateNoRemovedHttpServerInjection(
                                        resourceMethod);
                return resourceMethod;
        }

        @SuppressWarnings("ReferenceEquality")
        private static @NonNull Boolean isOptionsSplat(@NonNull ResourcePath resourcePath) {
                requireNonNull(resourcePath);
                return resourcePath == ResourcePath.OPTIONS_SPLAT_RESOURCE_PATH;
        }

        @NonNull
        private static Set<@NonNull HttpMethod> allowedHttpMethodsForResponse(@NonNull Map<@NonNull HttpMethod, @NonNull ResourceMethod> matchingResourceMethodsByHttpMethod,
                                                                                                                                                                                                                                                                                                @NonNull Boolean includeOptions, @NonNull ServerType serverType) {
                requireNonNull(matchingResourceMethodsByHttpMethod);
                requireNonNull(includeOptions);

                Set<HttpMethod> allowedHttpMethods = EnumSet.noneOf(HttpMethod.class);
                allowedHttpMethods.addAll(matchingResourceMethodsByHttpMethod.keySet());

                if (includeOptions)
                        allowedHttpMethods.add(HttpMethod.OPTIONS);

                if ((serverType == ServerType.HTTP && matchingResourceMethodsByHttpMethod.containsKey(HttpMethod.GET))
                                || matchingResourceMethodsByHttpMethod.containsKey(HttpMethod.HEAD))
                        allowedHttpMethods.add(HttpMethod.HEAD);

                return allowedHttpMethods;
        }

        @NonNull
        MarshaledResponse provideFailsafeMarshaledResponse(@NonNull Request request,
                                                                                                                                                                                                                                                         @NonNull Throwable throwable) {
                requireNonNull(request);
                requireNonNull(throwable);

                Integer statusCode = 500;
                Charset charset = StandardCharsets.UTF_8;

                return MarshaledResponse.withStatusCode(statusCode)
                                .headers(Map.of("Content-Type", List.of(format("text/plain; charset=%s", charset.name()))))
                                .body(format("HTTP %s: %s", statusCode, StatusCode.fromStatusCode(statusCode).get().getReasonPhrase()).getBytes(charset))
                                .build();
        }

        /**
         * Publishes shutdown intent, joins the lifecycle uninterruptibly, restores
         * the caller's interrupt status, and applies the terminal result.
         *
         * @throws SokletShutdownIncompleteException if complete termination cannot be proven
         * @throws SokletUnexpectedTerminationException if a lifecycle component terminated
         * unexpectedly after readiness
         * @throws IllegalStateException as a best-effort diagnostic when the current
         * thread is contributing to the unpublished shutdown barrier
         */
        void close() {
                shutdown();
                boolean interrupted = false;
                ShutdownResult result;

                try {
                        for (;;) {
                                try {
                                        result = awaitCoreShutdown();
                                        break;
                                } catch (InterruptedException exception) {
                                        interrupted = true;
                                }
                        }

                        this.directLifecycle.throwIfUnsuccessfulShutdown(result);
                } finally {
                        if (interrupted)
                                Thread.currentThread().interrupt();
                }
        }

        /**
         * Returns a racy diagnostic snapshot of this one-shot lifecycle.
         *
         * @return current lifecycle status
         */
        @NonNull
        SokletStatus getStatus() {
                return this.directLifecycle.publicStatus();
        }

        /**
         * Returns the immutable terminal result after it has been published.
         *
         * @return terminal result, otherwise empty
         */
        @NonNull
        Optional<@NonNull ShutdownResult> getShutdownResult() {
                return this.directLifecycle.publicResult();
        }

        void initializeForSimulator(@NonNull StartupContext startupContext,
                        @NonNull DeadlineWaiter waiter) {
                this.directLifecycle.initializeForSimulator(startupContext, waiter);
        }

        @NonNull
        SokletDirectLifecycle getDirectLifecycle() {
                return this.directLifecycle;
        }

        @NonNull
        SokletConfig getSokletConfig() {
                return this.sokletConfig;
        }

        /**
         * Returns the stable compatibility lock retained from earlier Soklet
         * versions.  The one-shot lifecycle does not use this caller-obtainable lock
         * for its internal state transitions, so holding it cannot delay lifecycle
         * progress.
         *
         * @return passive compatibility lock
         */
        @NonNull
        ReentrantLock getLock() {
                return this.lock;
        }

        /**
         * Returns the stable compatibility reference initially containing the latch
         * released when this one-shot lifecycle publishes its terminal result.
         * Soklet retains that original latch directly, so replacing the exposed
         * reference value cannot redirect lifecycle publication.
         *
         * @return one-shot shutdown-latch reference
         */
        @NonNull
        AtomicReference<@NonNull CountDownLatch>
                        getAwaitShutdownLatchReference() {
                return this.awaitShutdownLatchReference;
        }

        void releaseAwaitShutdownLatch() {
                this.awaitShutdownLatch.countDown();
        }

        @ThreadSafe
        static class DefaultSimulator implements Simulator {
                @Nullable
                private MockHttpServer server;
                @Nullable
                private MockSseServer sseServer;
                @NonNull
                private final SimulatorOptions simulatorOptions;
                private final @Nullable DefaultMcpServer mcpServer;
                private final @Nullable SimulatorScopeDispatchGate scopeDispatchGate;
                private final @Nullable Object executionOwnerToken;
                @NonNull
                private final Object mcpSimulationLock;
                @NonNull
                private final Object scopeStateLock;
                private volatile @Nullable SimulationSession mcpSimulationSession;
                private volatile @Nullable SimulationSession rejectedMcpSimulationSession;
                private volatile @Nullable StreamLifecycleCoordinator streamLifecycleCoordinator;
                private @Nullable Supplier<@NonNull StreamLifecycleCoordinator> streamLifecycleCoordinatorFactoryForTests;
                private volatile @Nullable StreamLifecycleCoordinator sseLifecycleCoordinator;
                private @Nullable Supplier<@NonNull StreamLifecycleCoordinator> sseLifecycleCoordinatorFactoryForTests;
                @NonNull
                private final AtomicBoolean closed;

                public DefaultSimulator(@Nullable MockHttpServer server,
                                                                                         @Nullable MockSseServer sseServer) {
                        this(server, sseServer, SimulatorOptions.defaultInstance(), null, null);
                }

                public DefaultSimulator(@Nullable MockHttpServer server,
                                                                                @Nullable MockSseServer sseServer,
                                                                                @NonNull SimulatorOptions simulatorOptions) {
                        this(server, sseServer, simulatorOptions, null, null);
                }

                DefaultSimulator(@Nullable MockHttpServer server,
                                @Nullable MockSseServer sseServer,
                                @NonNull SimulatorOptions simulatorOptions,
                                @Nullable DefaultMcpServer mcpServer) {
                        this(server, sseServer, simulatorOptions, mcpServer, null);
                }

                DefaultSimulator(@Nullable MockHttpServer server,
                                @Nullable MockSseServer sseServer,
                                @NonNull SimulatorOptions simulatorOptions,
                                @Nullable DefaultMcpServer mcpServer,
                                @Nullable SimulatorScopeDispatchGate scopeDispatchGate) {
                        this(server, sseServer, simulatorOptions, mcpServer, scopeDispatchGate, null);
                }

                DefaultSimulator(@Nullable MockHttpServer server, @Nullable MockSseServer sseServer,
                                @NonNull SimulatorOptions simulatorOptions, @Nullable DefaultMcpServer mcpServer,
                                @Nullable SimulatorScopeDispatchGate scopeDispatchGate, @Nullable Object executionOwnerToken) {
                        this.server = server;
                        this.sseServer = sseServer;
                        this.simulatorOptions = requireNonNull(simulatorOptions);
                        this.mcpServer = mcpServer;
                        this.scopeDispatchGate = scopeDispatchGate;
                        this.executionOwnerToken = executionOwnerToken;
                        this.mcpSimulationLock = new Object();
                        this.scopeStateLock = new Object();
                        this.closed = new AtomicBoolean();
                }

                private void runStreamCallback(@NonNull Runnable callback) {
                        Object token = this.executionOwnerToken;
                        try (LifecycleExecutionContext.Scope ignored = token == null ? null : LifecycleExecutionContext.enter(token)) {
                                callback.run();
                        }
                }

                @Override
                @NonNull
                public McpSimulation startMcpRequest(@NonNull Request request) {
                        return startMcpRequest(requireNonNull(request),
                                        McpSimulationOptions.defaultInstance());
                }

                @Override
                @NonNull
                public McpSimulation startMcpRequest(@NonNull Request request,
                                @NonNull McpSimulationOptions options) {
                        Runnable releaseScope = enterScope(InternalLifecycleComponentType.MCP);
                        try {
                                synchronized (this.mcpSimulationLock) {
                                        if (this.mcpServer == null)
                                                throw new IllegalStateException(
                                                                "You must specify a McpServer in your SokletConfig to simulate MCP requests");
                                        if (this.mcpSimulationSession == null
                                                        && this.scopeDispatchGate == null)
                                                this.mcpSimulationSession = this.mcpServer
                                                                .openSimulationSession();
                                        if (this.mcpSimulationSession == null)
                                                throw new IllegalStateException(
                                                                "The MCP simulator participant is not ready");
                                        return this.mcpSimulationSession.start(requireNonNull(request),
                                                        requireNonNull(options));
                                }
                        } finally {
                                releaseScope.run();
                        }
                }

                void openMcpScope() {
                        synchronized (this.mcpSimulationLock) {
                                requireScopeOpenWhileLocked();
                                if (this.mcpServer == null)
                                        throw new IllegalStateException(
                                                        "The simulator scope has no MCP participant");
                                if (this.mcpSimulationSession != null)
                                        throw new IllegalStateException(
                                                        "The MCP simulator participant was already started");
                                this.mcpServer.openSimulationSession(session -> {
                                        synchronized (this.scopeStateLock) {
                                                if (this.mcpSimulationSession != null
                                                                || this.rejectedMcpSimulationSession != null)
                                                        throw new IllegalStateException(
                                                                        "The MCP simulator participant was already started");
                                                SimulationSession exactSession = requireNonNull(session);
                                                if (this.closed.get()) {
                                                        // The runtime will reject and roll back this unclaimed session,
                                                        // but custom execution or registration cleanup may finish later.
                                                        // Retain its proof handle without exposing it as an active session.
                                                        this.rejectedMcpSimulationSession = exactSession;
                                                        requireScopeOpenWhileLocked();
                                                }
                                                this.mcpSimulationSession = exactSession;
                                        }
                                });
                        }
                }

                boolean sealScope() {
                        // Close dispatch admission without waiting for a startup call that may be
                        // inside the MCP session lock.  A request racing shutdown must lose at the
                        // owner gate, and startup cancellation must remain deadline-bounded.
                        synchronized (this.scopeStateLock) {
                                if (!this.closed.compareAndSet(false, true))
                                        return false;
                                if (this.scopeDispatchGate != null)
                                        this.scopeDispatchGate.seal();
                                return true;
                        }
                }

                void quiesceMcpScope() {
                        SimulationSession session = mcpSimulationSession();
                        if (session != null)
                                session.quiesce();
                }

                void quiesceHttpScope() {
                        StreamLifecycleCoordinator coordinator = this.streamLifecycleCoordinator;
                        if (coordinator != null)
                                coordinator.requestGracefulShutdown();
                }

                void forceHttpScope() {
                        StreamLifecycleCoordinator coordinator = this.streamLifecycleCoordinator;
                        if (coordinator != null)
                                coordinator.force();
                }

                boolean awaitHttpScopeTermination(long absoluteDeadlineNanos,
                                @NonNull NanoClock clock) throws InterruptedException {
                        StreamLifecycleCoordinator coordinator = this.streamLifecycleCoordinator;
                        if (coordinator == null)
                                return true;
                        long remaining = absoluteDeadlineNanos - requireNonNull(clock).nanoTime();
                        return coordinator.awaitTermination(System.nanoTime() + Math.max(0L, remaining));
                }

                boolean httpScopeTerminationProven() {
                        StreamLifecycleCoordinator coordinator = this.streamLifecycleCoordinator;
                        return coordinator == null || coordinator.isTerminated();
                }

                @NonNull
                Set<@NonNull InternalResidualActivityType> httpScopeResidualActivity() {
                        StreamLifecycleCoordinator coordinator = this.streamLifecycleCoordinator;
                        if (coordinator == null || coordinator.isTerminated())
                                return Set.of();
                        StreamLifecycleCoordinator.Snapshot snapshot = coordinator.snapshot();
                        EnumSet<InternalResidualActivityType> residual = EnumSet.noneOf(InternalResidualActivityType.class);
                        if (snapshot.reservations() > 0)
                                residual.add(InternalResidualActivityType.STREAM);
                        if (snapshot.queuedProducers() > 0 || snapshot.runningProducers() > 0 || snapshot.reservations() == 0)
                                residual.add(InternalResidualActivityType.EXECUTOR_TASK);
                        if (snapshot.callbacks() > 0 || snapshot.diagnostics() > 0 || snapshot.publisherLifetimes() > 0)
                                residual.add(InternalResidualActivityType.CALLBACK);
                        return Collections.unmodifiableSet(residual);
                }

                void setStreamLifecycleCoordinatorFactoryForTests(@NonNull Supplier<@NonNull StreamLifecycleCoordinator> factory) {
                        synchronized (this.scopeStateLock) {
                                requireScopeOpenWhileLocked();
                                if (this.streamLifecycleCoordinator != null)
                                        throw new IllegalStateException("Streaming lifecycle coordinator was already created");
                                this.streamLifecycleCoordinatorFactoryForTests = requireNonNull(factory);
                        }
                }

                @NonNull
                Optional<@NonNull StreamLifecycleCoordinator> getStreamLifecycleCoordinatorForTests() {
                        return Optional.ofNullable(this.streamLifecycleCoordinator);
                }

                private StreamLifecycleCoordinator.@Nullable Reservation reserveHttpStream() {
                        synchronized (this.scopeStateLock) {
                                if (this.closed.get())
                                        return null;
                                if (this.streamLifecycleCoordinator == null) {
                                        MockHttpServer server = requireNonNull(this.server);
                                        LifecycleObserver observer = server.getSokletConfig().orElseThrow().getAggregateLifecycleObserver();
                                        this.streamLifecycleCoordinator = this.streamLifecycleCoordinatorFactoryForTests == null
                                                        ? StreamLifecycleCoordinator.withCallbackExecution(server.streamingLifecycleCapacity,
                                                                        server.streamingCallbackConcurrency, server.streamingCleanupTimeout, throwable -> {
                                                                try {
                                                                        observer.didReceiveLogEvent(LogEvent.with(LogEventType.SERVER_INTERNAL_ERROR,
                                                                                        "A simulated streaming lifecycle supervision operation failed or exceeded its deadline")
                                                                                        .throwable(throwable).build());
                                                                } catch (Throwable observerFailure) {
                                                                        LifecycleObserverLogFallback.report(observerFailure);
                                                                }
                                                        }, this::runStreamCallback)
                                                        : requireNonNull(this.streamLifecycleCoordinatorFactoryForTests.get());
                                }
                                return this.streamLifecycleCoordinator.tryReserve();
                        }
                }

                void quiesceSseScope() {
                        MockSseServer server = this.sseServer;
                        if (server != null)
                                server.stop();
                        StreamLifecycleCoordinator coordinator = this.sseLifecycleCoordinator;
                        if (coordinator != null)
                                coordinator.force();
                }

                void forceSseScope() {
                        quiesceSseScope();
                }

                boolean awaitSseScopeTermination(long absoluteDeadlineNanos,
                                @NonNull NanoClock clock) throws InterruptedException {
                        StreamLifecycleCoordinator coordinator = this.sseLifecycleCoordinator;
                        return coordinator == null || coordinator.awaitTermination(System.nanoTime()
                                        + Math.max(0L, absoluteDeadlineNanos - requireNonNull(clock).nanoTime()));
                }

                boolean sseScopeTerminationProven() {
                        StreamLifecycleCoordinator coordinator = this.sseLifecycleCoordinator;
                        return coordinator == null || coordinator.isTerminated();
                }

                @NonNull
                Set<@NonNull InternalResidualActivityType> sseScopeResidualActivity() {
                        StreamLifecycleCoordinator coordinator = this.sseLifecycleCoordinator;
                        if (coordinator == null || coordinator.isTerminated())
                                return Set.of();
                        StreamLifecycleCoordinator.Snapshot snapshot = coordinator.snapshot();
                        EnumSet<InternalResidualActivityType> residual = EnumSet.noneOf(InternalResidualActivityType.class);
                        if (snapshot.reservations() > 0)
                                residual.add(InternalResidualActivityType.STREAM);
                        if (snapshot.queuedProducers() > 0 || snapshot.runningProducers() > 0 || snapshot.reservations() == 0)
                                residual.add(InternalResidualActivityType.EXECUTOR_TASK);
                        if (snapshot.callbacks() > 0 || snapshot.diagnostics() > 0 || snapshot.retainedWork() > 0)
                                residual.add(InternalResidualActivityType.CALLBACK);
                        return Collections.unmodifiableSet(residual);
                }

                void setSseLifecycleCoordinatorFactoryForTests(@NonNull Supplier<@NonNull StreamLifecycleCoordinator> factory) {
                        synchronized (this.scopeStateLock) {
                                requireScopeOpenWhileLocked();
                                if (this.sseLifecycleCoordinator != null)
                                        throw new IllegalStateException("SSE lifecycle coordinator was already created");
                                this.sseLifecycleCoordinatorFactoryForTests = requireNonNull(factory);
                        }
                }

                @NonNull
                Optional<@NonNull StreamLifecycleCoordinator> getSseLifecycleCoordinatorForTests() {
                        return Optional.ofNullable(this.sseLifecycleCoordinator);
                }

                private StreamLifecycleCoordinator.@Nullable Reservation reserveSseStream() {
                        synchronized (this.scopeStateLock) {
                                if (this.closed.get())
                                        return null;
                                return sseCoordinatorWhileLocked().tryReserve();
                        }
                }

                private @NonNull StreamLifecycleCoordinator sseCoordinatorWhileLocked() {
                        if (this.sseLifecycleCoordinator == null) {
                                MockSseServer server = requireNonNull(this.sseServer);
                                LifecycleObserver observer = server.getSokletConfig().orElseThrow().getAggregateLifecycleObserver();
                                this.sseLifecycleCoordinator = this.sseLifecycleCoordinatorFactoryForTests == null
                                                ? StreamLifecycleCoordinator.withCallbackExecution(server.streamingLifecycleCapacity,
                                                                DefaultSseServer.STREAMING_COORDINATOR_CALLBACK_CONCURRENCY,
                                                                DefaultSseServer.STREAMING_COORDINATOR_CLEANUP_GRACE, throwable -> {
                                                        try {
                                                                observer.didReceiveLogEvent(LogEvent.with(LogEventType.SSE_SERVER_INTERNAL_ERROR,
                                                                                "A simulated SSE cleanup operation failed or exceeded its deadline")
                                                                                .throwable(throwable).build());
                                                        } catch (Throwable observerFailure) {
                                                                LifecycleObserverLogFallback.report(observerFailure);
                                                        }
                                                }, this::runStreamCallback) : requireNonNull(this.sseLifecycleCoordinatorFactoryForTests.get());
                        }
                        return this.sseLifecycleCoordinator;
                }

                void forceMcpScope() {
                        SimulationSession session = mcpSimulationSession();
                        if (session != null)
                                session.force();
                }

                boolean awaitMcpScopeTermination(long absoluteDeadlineNanos,
                                @NonNull NanoClock clock) throws InterruptedException {
                        SimulationSession session = mcpEvidenceSession();
                        return session == null
                                        || session.awaitTermination(absoluteDeadlineNanos,
                                                        requireNonNull(clock)::nanoTime);
                }

                boolean mcpScopeTerminationProven() {
                        SimulationSession session = mcpEvidenceSession();
                        return session == null || session.terminationProven();
                }

                @NonNull
                Set<@NonNull InternalResidualActivityType> mcpScopeResidualActivity() {
                        SimulationSession session = mcpEvidenceSession();
                        if (session == null)
                                return Set.of();
                        com.soklet.internal.mcp.protocol.McpServerRuntimeBridge.LifecycleEvidence
                                        evidence = session.lifecycleEvidence();
                        EnumSet<InternalResidualActivityType> residual =
                                        EnumSet.noneOf(InternalResidualActivityType.class);
                        if (evidence.executorTask() || evidence.subscriptionRegistration())
                                residual.add(InternalResidualActivityType.EXECUTOR_TASK);
                        if (evidence.stream())
                                residual.add(InternalResidualActivityType.STREAM);
                        if (evidence.callback())
                                residual.add(InternalResidualActivityType.CALLBACK);
                        return Collections.unmodifiableSet(residual);
                }

                void releaseMcpScopeEvidence() {
                        SimulationSession activeSession = mcpSimulationSession();
                        SimulationSession rejectedSession = rejectedMcpSimulationSession();
                        if (activeSession != null)
                                activeSession.releaseLifecycleEvidence();
                        if (rejectedSession != null && rejectedSession != activeSession)
                                rejectedSession.releaseLifecycleEvidence();
                        clearMcpSimulationSession(activeSession);
                        clearRejectedMcpSimulationSession(rejectedSession);
                }

                void releaseHttpAndSseScopeState() {
                        MockHttpServer httpServer = this.server;
                        MockSseServer serverSentEventsServer = this.sseServer;
                        if (httpServer != null)
                                httpServer.releaseSimulationScopeState();
                        if (serverSentEventsServer != null)
                                serverSentEventsServer.releaseSimulationScopeState();
                }

                @Nullable
                private SimulationSession mcpSimulationSession() {
                        return this.mcpSimulationSession;
                }

                @Nullable
                private SimulationSession rejectedMcpSimulationSession() {
                        return this.rejectedMcpSimulationSession;
                }

                @Nullable
                private SimulationSession mcpEvidenceSession() {
                        SimulationSession activeSession = mcpSimulationSession();
                        return activeSession == null
                                        ? rejectedMcpSimulationSession() : activeSession;
                }

                private void clearMcpSimulationSession(@Nullable SimulationSession session) {
                        synchronized (this.mcpSimulationLock) {
                                if (this.mcpSimulationSession == session)
                                        this.mcpSimulationSession = null;
                        }
                }

                private void clearRejectedMcpSimulationSession(
                                @Nullable SimulationSession session) {
                        synchronized (this.scopeStateLock) {
                                if (this.rejectedMcpSimulationSession == session)
                                        this.rejectedMcpSimulationSession = null;
                        }
                }

                @NonNull
                @Override
                public HttpRequestResult performHttpRequest(@NonNull Request request) {
                        Runnable releaseScope = enterScope(InternalLifecycleComponentType.HTTP);
                        try (HttpTransportResponseFeedback feedback = HttpTransportResponseFeedback.create()) {
                                // A caller may concurrently reuse an immutable Request. Each simulated
                                // dispatch needs its own identity, shared by handling and stream callbacks.
                                return performHttpRequestWhileAdmitted(requireNonNull(request).copy().finish(), feedback);
                        } finally {
                                releaseScope.run();
                        }
                }

                @NonNull
                private HttpRequestResult performHttpRequestWhileAdmitted(
                                @NonNull Request request, @NonNull HttpTransportResponseFeedback feedback) {
                        MockHttpServer server = this.server;

                        if (server == null)
                                throw new IllegalStateException(format("You must specify a %s in your %s to simulate requests",
                                                HttpServer.class.getSimpleName(), SokletConfig.class.getSimpleName()));

                        AtomicReference<@Nullable HttpRequestResult> requestResultHolder = new AtomicReference<>();
                        HttpServer.RequestHandler requestHandler = server.getRequestHandler().orElse(null);

                        if (requestHandler == null)
                                throw new IllegalStateException("You must register a request handler prior to simulating requests");

                        AtomicReference<StreamLifecycleCoordinator.@Nullable Reservation> reservationHolder = new AtomicReference<>();
                        Throwable primaryFailure = null;
                        Throwable terminalCleanupFailure = null;
                        HttpRequestResult completedResult;
                        try {
                                requestHandler.handleRequest(request, HttpTransportResponseFeedback.fromWriter(feedback, requestResult -> {
                                        // Simulated responses do not run transport compression or retain its private HEAD input.
                                        HttpRequestResult result = projectSimulatedHttpResponse(request, requestResult);
                                        if (result.getMarshaledResponse().isStreaming()) {
                                                Instant streamEstablishedAt = Instant.now();
                                                long streamEstablishedNanos = System.nanoTime();
                                                StreamLifecycleCoordinator.Reservation reservation = reserveHttpStream();
                                                if (reservation == null) {
                                                        result.getMarshaledResponse().close();
                                                        MarshaledResponse unavailable = DefaultHttpServer.provideDefaultStreamingAdmissionRejection();
                                                        requestResultHolder.set(result.copy().marshaledResponse(unavailable).response(null).finish());
                                                        Throwable rejection = new RejectedExecutionException(
                                                                        "Streaming lifecycle capacity is unavailable.");
                                                        StreamLifecycleCoordinator coordinator = this.streamLifecycleCoordinator;
                                                        StreamTerminationReason reason = this.closed.get() || coordinator == null || !coordinator.snapshot().accepting()
                                                                        ? StreamTerminationReason.SERVER_STOPPING : StreamTerminationReason.BACKPRESSURE;
                                                        Runnable notification = () -> notifyDidTerminateSimulatorResponseStream(request, result, Instant.now(),
                                                                        Duration.ZERO, reason, rejection);
                                                        if ((coordinator == null || !coordinator.dispatchRejectionObserver(notification))
                                                                        && reason != StreamTerminationReason.SERVER_STOPPING) {
                                                                try {
                                                                        server.getSokletConfig().orElseThrow().getAggregateLifecycleObserver().didReceiveLogEvent(
                                                                                                LogEvent.with(LogEventType.RESPONSE_STREAM_CANCELED,
                                                                                                        "Unadmitted stream rejection observer capacity was unavailable.").request(request).build());
                                                                } catch (Throwable observerFailure) {
                                                                        LifecycleObserverLogFallback.report(observerFailure);
                                                                }
                                                        }
                                                        // Synchronous replacement keeps didWrite/finish callbacks and metrics finite.
                                                        return HttpTransportResponseOutcome.replaced(unavailable, rejection, null);
                                                }
                                                reservationHolder.set(reservation);
                                                HttpResponseStreamObservation observation = result.getResponseStreamObservation();
                                                if (observation != null) {
                                                        SokletConfig config = server.getSokletConfig().orElseThrow();
                                                        LifecycleObserver observer = config.getAggregateLifecycleObserver();
                                                        observation.prepare(new DefaultStreamingResponseHandle(ServerType.HTTP, observation.getRequest(),
                                                                        result.getResourceMethod().orElse(null), result.getMarshaledResponse(), streamEstablishedAt),
                                                                        streamEstablishedNanos, config.getMetricsCollector(), observer, logEvent -> {
                                                                try { observer.didReceiveLogEvent(logEvent); }
                                                                catch (Throwable observerFailure) { LifecycleObserverLogFallback.report(observerFailure); }
                                                        });
                                                }
                                        }
                                        requestResultHolder.set(result);
                                        return HttpTransportResponseOutcome.accepted();
                                }));
                                completedResult = materializeStreamingResponse(request, requestResultHolder.get(), reservationHolder.get());
                                if (!completedResult.getMarshaledResponse().isStreaming())
                                        completedResult = materializeFiniteSimulationResult(completedResult);
                        } catch (RuntimeException | Error failure) {
                                primaryFailure = failure;
                                throw failure;
                        } finally {
                                // Also retire a reservation if handling fails before inline producer entry.
                                StreamLifecycleCoordinator.Reservation reservation = reservationHolder.get();
                                try {
                                        HttpRequestResult result = requestResultHolder.get();
                                        if (result != null) {
                                                try { result.getMarshaledResponse().close(); }
                                                catch (RuntimeException | Error cleanupFailure) {
                                                        if (primaryFailure != null && !sameInstance(primaryFailure, cleanupFailure)) primaryFailure.addSuppressed(cleanupFailure);
                                                        if (reservation != null) reservation.reportCleanupFailure(cleanupFailure);
                                                        else if (primaryFailure == null) terminalCleanupFailure = cleanupFailure;
                                                }
                                        }
                                } finally {
                                        if (reservation != null) reservation.complete();
                                }
                        }
                        if (terminalCleanupFailure instanceof RuntimeException runtime) throw runtime;
                        if (terminalCleanupFailure instanceof Error error) throw error;
                        return completedResult;
                }

                private static @NonNull HttpRequestResult materializeFiniteSimulationResult(@NonNull HttpRequestResult result) {
                        MarshaledResponse response = result.getMarshaledResponse();
                        Throwable primary = null;
                        try {
                                byte[] bytes = response.bodyBytesOrNull();
                                MarshaledResponse snapshot = MarshaledResponse.withStatusCode(response.getStatusCode())
                                                .headers(response.getHeaders()).cookies(response.getCookies()).body(bytes).build();
                                return result.copy().marshaledResponse(snapshot).finish();
                        } catch (RuntimeException | Error failure) {
                                primary = failure;
                                throw failure;
                        } finally {
                                MarshaledResponse.closeChannels(response.ownedChannels(), primary);
                        }
                }

                private @NonNull HttpRequestResult materializeAcceptedSseSimulationResult(@NonNull HttpRequestResult result) {
                        MarshaledResponse response = result.getMarshaledResponse();
                        Object token = this.executionOwnerToken;
                        try (LifecycleExecutionContext.Scope ignored = token == null ? null : LifecycleExecutionContext.enter(token)) {
                                Throwable primary = null;
                                try {
                                        // Accepted handshakes have no body. Keep their metadata without retaining
                                        // channel ownership after the headers have been materialized.
                                        MarshaledResponse snapshot = MarshaledResponse.withStatusCode(response.getStatusCode())
                                                        .headers(response.getHeaders()).cookies(response.getCookies()).build();
                                        return result.copy().marshaledResponse(snapshot).finish();
                                } catch (RuntimeException | Error failure) {
                                        primary = failure;
                                        throw failure;
                                } finally {
                                        MarshaledResponse.closeChannels(response.ownedChannels(), primary);
                                }
                        }
                }

                private static @NonNull HttpRequestResult projectSimulatedHttpResponse(@NonNull Request request, @NonNull HttpRequestResult result) {
                        MarshaledResponse response = normalizeBodylessMarshaledResponse(request, result.getMarshaledResponse());
                        boolean head = request.getHttpMethod() == HttpMethod.HEAD;
                        if (!response.isStreaming())
                                response = response.copy().headers(DefaultHttpServer.simulatedFiniteResponseHeaders(request, response,
                                                result.getHeadResponseBodyLength().orElse(null))).finish();

                        // The transport suppresses HEAD bytes even when an interceptor or custom
                        // HeadHandler leaves a body. Never execute a suppressed stream producer.
                        if (head)
                                response = response.copy().withoutBody().withoutStreamingResponseBody().finish();
                        return result.copy().marshaledResponse(response).headResponseCompressionBody(null).finish();
                }

                private static @NonNull HttpRequestResult projectSimulatedSseRejection(@NonNull Request request, @NonNull HttpRequestResult result) {
                        MarshaledResponse response = normalizeBodylessMarshaledResponse(request, result.getMarshaledResponse());
                        boolean head = request.getHttpMethod() == HttpMethod.HEAD;
                        if (head)
                                response = response.copy().withoutBody().withoutStreamingResponseBody().finish();
                        int status = response.getStatusCode();
                        long length = response.getBodyLength();
                        Set<String> transportHeaders = new HashSet<>(Set.of("connection", "keep-alive", "content-length",
                                        "transfer-encoding", "te", "trailer", "upgrade", "proxy-connection"));
                        for (String value : response.getHeaders().getOrDefault("Connection", List.of()))
                                for (String name : value.split(",", -1))
                                        transportHeaders.add(name.trim().toLowerCase(Locale.ROOT));
                        response = response.copy().headers(headers -> {
                                headers.keySet().removeIf(name -> transportHeaders.contains(name.toLowerCase(Locale.ROOT)));
                                if (!head && status != 204 && status != 304)
                                        headers.put("Content-Length", List.of(Long.toString(length)));
                                headers.put("Connection", List.of("close"));
                                headers.putIfAbsent("Date", List.of(HttpDate.currentSecondHeaderValue()));
                        }).finish();
                        return result.copy().marshaledResponse(response).headResponseCompressionBody(null).headResponseBodyLength(null).finish();
                }

                @NonNull
                private HttpRequestResult materializeStreamingResponse(@NonNull Request request,
                                @Nullable HttpRequestResult requestResult, StreamLifecycleCoordinator.@Nullable Reservation reservedLifetime) {
                        requireNonNull(request);

                        if (requestResult == null)
                                throw new IllegalStateException("No HTTP request result was produced by the simulator");

                        StreamingResponseBody stream = requestResult.getMarshaledResponse().getStreamingResponseBody().orElse(null);

                        if (stream == null)
                                return requestResult;

                        StreamLifecycleCoordinator.Reservation reservation = requireNonNull(reservedLifetime);
                        AtomicReference<@Nullable HttpRequestResult> result = new AtomicReference<>();
                        AtomicReference<@Nullable Throwable> failure = new AtomicReference<>();
                        LifecycleObserver cleanupObserver = requireNonNull(this.server).getSokletConfig().orElseThrow()
                                        .getAggregateLifecycleObserver();
                        reservation.bindCleanupFailureObserver(throwable -> {
                                try {
                                        cleanupObserver.didReceiveLogEvent(LogEvent.with(LogEventType.RESPONSE_STREAM_CLOSE_FAILED,
                                                                "A simulated streaming response cleanup operation failed")
                                                        .throwable(throwable).request(request)
                                                        .resourceMethod(requestResult.getResourceMethod().orElse(null))
                                                        .marshaledResponse(requestResult.getMarshaledResponse()).build());
                                } catch (Throwable observerFailure) {
                                        LifecycleObserverLogFallback.report(observerFailure);
                                }
                        });
                        boolean entered = reservation.executeInline(() -> {
                                try {
                                        result.set(materializeStreamingResponseWhileReserved(request, requestResult, stream, reservation));
                                } catch (Throwable throwable) {
                                        failure.set(throwable);
                                } finally {
                                        reservation.complete();
                                }
                        });
                        if (!entered) {
                                StreamTerminationReason reason = reservation.reason().orElse(StreamTerminationReason.SERVER_STOPPING);
                                Throwable cause = reservation.cause().orElse(null);
                                try {
                                        notifyDidTerminateSimulatorResponseStream(reservation, request, requestResult, Instant.now(),
                                                        Duration.ZERO, reason, cause);
                                } finally {
                                        reservation.complete();
                                }
                                throw new IllegalStateException("Simulated streaming response was canceled: " + reason.name(),
                                                new StreamingResponseCanceledException(reason, cause));
                        }
                        Throwable throwable = failure.get();
                        if (throwable instanceof RuntimeException exception)
                                throw exception;
                        if (throwable instanceof Error error)
                                throw error;
                        if (throwable != null)
                                throw new IllegalStateException("Simulated streaming response failed.", throwable);
                        return requireNonNull(result.get());
                }

                @NonNull
                private HttpRequestResult materializeStreamingResponseWhileReserved(@NonNull Request request,
                                @NonNull HttpRequestResult requestResult, @NonNull StreamingResponseBody stream,
                                StreamLifecycleCoordinator.@NonNull Reservation reservation) {

                        byte[] bytes;
                        java.util.concurrent.atomic.AtomicLong acceptedBytes = new java.util.concurrent.atomic.AtomicLong();
                        Instant streamStarted = Instant.now();

                        try {
                                bytes = materializeStreamingResponseBody(request, requestResult, stream, reservation, acceptedBytes);
                                if (!reservation.completeTransport())
                                        throw new StreamingResponseCanceledException(reservation.reason().orElse(StreamTerminationReason.SERVER_STOPPING),
                                                        reservation.cause().orElse(null));
                                notifyDidTerminateSimulatorResponseStream(reservation, request, requestResult, streamStarted,
                                                Duration.between(streamStarted, Instant.now()), null, null, acceptedBytes.get());
                        } catch (StreamingResponseCanceledException e) {
                                StreamTerminationReason cancelationReason = e.getCancelationReason();
                                Throwable cause = e.getCancelationCause().orElse(null);
                                notifyDidTerminateSimulatorResponseStream(reservation, request, requestResult, streamStarted,
                                                Duration.between(streamStarted, Instant.now()), cancelationReason, cause, acceptedBytes.get());
                                throw new IllegalStateException("Simulated streaming response was canceled: " + cancelationReason.name(), e);
                        } catch (InterruptedException e) {
                                Thread.currentThread().interrupt();
                                notifyDidTerminateSimulatorResponseStream(reservation, request, requestResult, streamStarted,
                                                Duration.between(streamStarted, Instant.now()), StreamTerminationReason.APPLICATION_CANCELED, e, acceptedBytes.get());
                                throw new IllegalStateException("Simulated streaming response was canceled: APPLICATION_CANCELED", e);
                        } catch (Throwable t) {
                                notifyDidTerminateSimulatorResponseStream(reservation, request, requestResult, streamStarted,
                                                Duration.between(streamStarted, Instant.now()), StreamTerminationReason.PRODUCER_FAILED, t, acceptedBytes.get());

                                if (t instanceof Error error)
                                        throw error;

                                throw new IllegalStateException("Simulated streaming response failed.", t);
                        }

                        try { requestResult.getMarshaledResponse().close(); }
                        catch (RuntimeException | Error cleanupFailure) { reservation.reportCleanupFailure(cleanupFailure); }
                        MarshaledResponse marshaledResponse = requestResult.getMarshaledResponse().copy()
                                        .withoutStreamingResponseBody()
                                        .body(bytes)
                                        .finish();

                        return requestResult.copy()
                                        .marshaledResponse(marshaledResponse)
                                        .finish();
                }

                private void notifyDidTerminateSimulatorResponseStream(StreamLifecycleCoordinator.@NonNull Reservation reservation,
                                @NonNull Request request, @NonNull HttpRequestResult requestResult,
                                @NonNull Instant establishedAt, @NonNull Duration streamDuration,
                                @Nullable StreamTerminationReason cancelationReason, @Nullable Throwable throwable) {
                        notifyDidTerminateSimulatorResponseStream(reservation, request, requestResult, establishedAt,
                                        streamDuration, cancelationReason, throwable, 0L);
                }

                private void notifyDidTerminateSimulatorResponseStream(StreamLifecycleCoordinator.@NonNull Reservation reservation,
                                @NonNull Request request, @NonNull HttpRequestResult requestResult,
                                @NonNull Instant establishedAt, @NonNull Duration streamDuration,
                                @Nullable StreamTerminationReason cancelationReason, @Nullable Throwable throwable, long bodyBytes) {
                        long terminalNanos = reservation.transportTerminatedNanos();
                        CountDownLatch delivered = new CountDownLatch(1);
                        reservation.dispatchTermination(() -> {
                                try {
                                        notifyDidTerminateSimulatorResponseStream(request, requestResult, establishedAt,
                                                        streamDuration, cancelationReason, throwable, terminalNanos, bodyBytes);
                                } finally {
                                        delivered.countDown();
                                }
                        });
                        boolean interrupted = false;
                        try {
                                for (;;) {
                                        try {
                                                delivered.await();
                                                return;
                                        } catch (InterruptedException ignored) {
                                                interrupted = true;
                                        }
                                }
                        } finally {
                                if (interrupted)
                                        Thread.currentThread().interrupt();
                        }
                }

                private void notifyDidTerminateSimulatorResponseStream(@NonNull Request request,
                                                                                                                                                                                                                                         @NonNull HttpRequestResult requestResult,
                                                                                                                                                                                                                                         @NonNull Instant establishedAt,
                                                                                                                                                                                                                                         @NonNull Duration streamDuration,
                                                                                                                                                                                                                                         @Nullable StreamTerminationReason cancelationReason,
                                                                                                                                                                                                                                         @Nullable Throwable throwable) {
                        notifyDidTerminateSimulatorResponseStream(request, requestResult, establishedAt, streamDuration,
                                        cancelationReason, throwable, 0L, 0L);
                }

                private void notifyDidTerminateSimulatorResponseStream(@NonNull Request request,
                                                                                                                                                                                                                                         @NonNull HttpRequestResult requestResult,
                                                                                                                                                                                                                                         @NonNull Instant establishedAt,
                                                                                                                                                                                                                                         @NonNull Duration streamDuration,
                                                                                                                                                                                                                                         @Nullable StreamTerminationReason cancelationReason,
                                                                                                                                                                                                                                         @Nullable Throwable throwable, long terminalNanos, long bodyBytes) {
                        requireNonNull(request);
                        requireNonNull(requestResult);
                        requireNonNull(establishedAt);
                        requireNonNull(streamDuration);

                        MockHttpServer server = this.server;
                        SokletConfig sokletConfig = server == null ? null : server.getSokletConfig().orElse(null);

                        if (sokletConfig == null)
                                return;

                        MarshaledResponse marshaledResponse = requestResult.getMarshaledResponse();
                        ResourceMethod resourceMethod = requestResult.getResourceMethod().orElse(null);
                        LifecycleObserver lifecycleObserver = sokletConfig.getAggregateLifecycleObserver();
                        HttpResponseStreamObservation observation = requestResult.getResponseStreamObservation();
                        StreamingResponseHandle streamingResponse = new DefaultStreamingResponseHandle(ServerType.HTTP,
                                        observation == null ? request : observation.getRequest(), resourceMethod, marshaledResponse, establishedAt);
                        StreamTermination termination = StreamTermination
                                        .with(cancelationReason == null ? StreamTerminationReason.COMPLETED : cancelationReason, streamDuration)
                                        .cause(throwable)
                                        .build();

                        if (observation != null && terminalNanos != 0L) {
                                HttpResponseStreamObservation.Delivery delivery = observation.deliver(streamingResponse, termination, terminalNanos, bodyBytes,
                                                sokletConfig.getMetricsCollector(), logEvent -> {
                                        try { lifecycleObserver.didReceiveLogEvent(logEvent); }
                                        catch (Throwable failure) { LifecycleObserverLogFallback.report(failure); }
                                });
                                streamingResponse = delivery.handle();
                                termination = delivery.termination();
                        }
                        if (cancelationReason != null && cancelationReason != StreamTerminationReason.CLIENT_DISCONNECTED
                                        && cancelationReason != StreamTerminationReason.SERVER_STOPPING) {
                                LogEventType type = cancelationReason == StreamTerminationReason.PRODUCER_FAILED
                                                ? LogEventType.RESPONSE_STREAM_FAILED : LogEventType.RESPONSE_STREAM_CANCELED;
                                try {
                                        lifecycleObserver.didReceiveLogEvent(LogEvent.with(type,
                                                        format("Streaming response terminated: %s", cancelationReason.name()))
                                                .throwable(throwable).request(request).resourceMethod(resourceMethod)
                                                .marshaledResponse(marshaledResponse).build());
                                } catch (Throwable observerFailure) {
                                        LifecycleObserverLogFallback.report(observerFailure);
                                }
                        }

                        try {
                                lifecycleObserver.willTerminateResponseStream(streamingResponse, termination);
                        } catch (Throwable t) {
                                try {
                                        lifecycleObserver.didReceiveLogEvent(LogEvent.with(LogEventType.LIFECYCLE_OBSERVER_WILL_TERMINATE_RESPONSE_STREAM_FAILED,
                                                                        format("An exception occurred while invoking %s::willTerminateResponseStream", LifecycleObserver.class.getSimpleName()))
                                                        .throwable(t)
                                                        .request(request)
                                                        .resourceMethod(resourceMethod)
                                                        .marshaledResponse(marshaledResponse)
                                                        .build());
                                } catch (Throwable observerFailure) {
                                        LifecycleObserverLogFallback.report(observerFailure);
                                }
                        }

                        try {
                                lifecycleObserver.didTerminateResponseStream(streamingResponse, termination);
                        } catch (Throwable t) {
                                try {
                                        lifecycleObserver.didReceiveLogEvent(LogEvent.with(LogEventType.LIFECYCLE_OBSERVER_DID_TERMINATE_RESPONSE_STREAM_FAILED,
                                                                        format("An exception occurred while invoking %s::didTerminateResponseStream", LifecycleObserver.class.getSimpleName()))
                                                        .throwable(t)
                                                        .request(request)
                                                        .resourceMethod(resourceMethod)
                                                        .marshaledResponse(marshaledResponse)
                                                        .build());
                                } catch (Throwable observerFailure) {
                                        LifecycleObserverLogFallback.report(observerFailure);
                                }
                        }
                }

                private void notifyDidReceiveSimulatorStreamCancelationCallbackFailure(@NonNull Request request,
                                                                                                                                                                                                                                                                                                        @NonNull HttpRequestResult requestResult,
                                                                                                                                                                                                                                                                                                        @NonNull Throwable throwable) {
                        requireNonNull(request);
                        requireNonNull(requestResult);
                        requireNonNull(throwable);

                        MockHttpServer server = this.server;
                        SokletConfig sokletConfig = server == null ? null : server.getSokletConfig().orElse(null);

                        if (sokletConfig == null)
                                return;

                        LifecycleObserver lifecycleObserver = sokletConfig.getAggregateLifecycleObserver();
                        ResourceMethod resourceMethod = requestResult.getResourceMethod().orElse(null);
                        MarshaledResponse marshaledResponse = requestResult.getMarshaledResponse();

                        try {
                                lifecycleObserver.didReceiveLogEvent(LogEvent.with(LogEventType.RESPONSE_STREAM_CANCELATION_CALLBACK_FAILED,
                                                                "An exception occurred while invoking a streaming response cancelation callback")
                                                .throwable(throwable)
                                                .request(request)
                                                .resourceMethod(resourceMethod)
                                                .marshaledResponse(marshaledResponse)
                                                .build());
                        } catch (Throwable observerFailure) {
                                LifecycleObserverLogFallback.report(observerFailure);
                        }
                }

                private byte @NonNull [] materializeStreamingResponseBody(@NonNull Request request,
                                                                                                                                                                                                                @NonNull HttpRequestResult requestResult,
                                @NonNull StreamingResponseBody stream,
                                StreamLifecycleCoordinator.@NonNull Reservation reservation, java.util.concurrent.atomic.@NonNull AtomicLong acceptedBytes) throws Exception {
                        requireNonNull(request);
                        requireNonNull(requestResult);
                        requireNonNull(stream);

                        Consumer<Throwable> cleanupFailureConsumer = throwable ->
                                        notifyDidReceiveSimulatorStreamCancelationCallbackFailure(request, requestResult, throwable);
                        SimulatorCancelationToken cancelationToken = new SimulatorCancelationToken(cleanupFailureConsumer, reservation);
                        reservation.bindTermination(cancelationToken::deliverCancelation);
                        SimulatorResponseOutput output = new SimulatorResponseOutput(cancelationToken,
                                        getSimulatorOptions().getStreamingResponseBodyLimitInBytes(), acceptedBytes);

                        try {
                                cancelationToken.throwIfCanceled();
                                if (stream instanceof StreamingResponseBody.PublisherBody publisherBody) {
                                        materializePublisher(publisherBody, output, cancelationToken, reservation);
                                } else {
                                        ManagedResponseStream managedResponseStream = new ManagedResponseStream(request, cancelationToken,
                                                        reservation::isGracefulShutdownRequested, null, null, output, reservation::beginCleanup,
                                                        throwable -> cancelSimulatorStream(cancelationToken, throwable), reservation::reportCleanupFailure);
                                        managedResponseStream.run(responseStream -> {
                                                if (stream instanceof StreamingResponseBody.WriterBody writerBody) {
                                                        writerBody.getWriter().writeTo(responseStream);
                                                } else if (stream instanceof StreamingResponseBody.InputStreamBody inputStreamBody) {
                                                        java.io.InputStream inputStream = responseStream.open(inputStreamBody.getInputStreamFactory());
                                                        byte[] buffer = new byte[inputStreamBody.getBufferSizeInBytes()];
                                                        int read;

                                                        while ((read = inputStream.read(buffer)) >= 0) {
                                                                responseStream.getCancelationToken().throwIfCanceled();
                                                                if (read > 0)
                                                                        responseStream.write(ByteBuffer.wrap(buffer, 0, read));
                                                        }
                                                } else if (stream instanceof StreamingResponseBody.ReaderBody readerBody) {
                                                        Reader reader = responseStream.open(readerBody.getReaderFactory());
                                                        materializeReader(readerBody, reader, responseStream);
                                                } else {
                                                        throw new IllegalStateException(format("Unsupported streaming response body type: %s", stream.getClass().getName()));
                                                }
                                        });
                                }
                        } catch (Throwable t) {
                                cancelSimulatorStream(cancelationToken, t);
                                StreamTerminationReason reason = cancelationToken.getCancelationReason()
                                                .orElse(StreamTerminationReason.PRODUCER_FAILED);
                                if (reason != StreamTerminationReason.PRODUCER_FAILED
                                                && !(t instanceof InterruptedException && reason == StreamTerminationReason.APPLICATION_CANCELED)
                                                && !(t instanceof StreamingResponseCanceledException canceledException
                                                && canceledException.getCancelationReason() == reason)) {
                                        Throwable cause = cancelationToken.getCancelationCause().orElse(null);
                                        StreamingResponseCanceledException canceledException = new StreamingResponseCanceledException(reason, cause);
                                        if (!sameInstance(t, cause) && !(t instanceof InterruptedException))
                                                canceledException.addSuppressed(t);
                                        throw canceledException;
                                }

                                if (t instanceof Exception exception)
                                        throw exception;

                                if (t instanceof Error error)
                                        throw error;

                                throw new RuntimeException(t);
                        }

                        byte[] bytes = output.toByteArray();
                        if (!reservation.completeProduction())
                                cancelationToken.throwIfCanceled();
                        cancelationToken.complete();
                        return bytes;
                }

                private void cancelSimulatorStream(@NonNull SimulatorCancelationToken cancelationToken,
                                @NonNull Throwable throwable) {
                        if (throwable instanceof StreamingResponseCanceledException canceledException) {
                                cancelationToken.cancel(canceledException.getCancelationReason(), canceledException.getCancelationCause().orElse(null));
                        } else if (throwable instanceof InterruptedException) {
                                Thread.currentThread().interrupt();
                                cancelationToken.cancel(cancelationToken.getCancelationReason()
                                                .orElse(StreamTerminationReason.APPLICATION_CANCELED), throwable);
                        } else {
                                cancelationToken.cancel(StreamTerminationReason.PRODUCER_FAILED, throwable);
                        }
                }

                private void materializeReader(com.soklet.StreamingResponseBody.@NonNull ReaderBody readerBody,
                                                                                                                                         @NonNull Reader reader,
                                                                                                                                         @NonNull ResponseStream responseStream) throws IOException, InterruptedException, StreamingResponseCanceledException, CharacterCodingException {
                        requireNonNull(readerBody);
                        requireNonNull(reader);
                        requireNonNull(responseStream);

                        CharsetEncoder encoder = readerBody.newEncoder();
                        int readSize = readerBody.getBufferSizeInCharacters();
                        // Preserve single-character reads while leaving room for a carried high surrogate.
                        CharBuffer charBuffer = CharBuffer.allocate(Math.max(2, readSize));
                        ByteBuffer byteBuffer = ByteBuffer.allocate(Math.max(128, (int) Math.ceil(readSize * encoder.maxBytesPerChar())));

                        while (true) {
                                charBuffer.limit(charBuffer.position() + Math.min(readSize, charBuffer.remaining()));
                                int read = reader.read(charBuffer);
                                charBuffer.limit(charBuffer.capacity());
                                if (read < 0)
                                        break;
                                responseStream.getCancelationToken().throwIfCanceled();
                                charBuffer.flip();
                                encodeCharsForSimulator(encoder, charBuffer, byteBuffer, false, responseStream);
                                charBuffer.compact();
                        }

                        charBuffer.flip();
                        encodeCharsForSimulator(encoder, charBuffer, byteBuffer, true, responseStream);

                        CoderResult result;
                        do {
                                result = encoder.flush(byteBuffer);
                                writeEncodedBytesForSimulator(byteBuffer, responseStream);
                                if (result.isError())
                                        result.throwException();
                        } while (result.isOverflow());
                }

                private void encodeCharsForSimulator(@NonNull CharsetEncoder encoder,
                                                                                                                                                                 @NonNull CharBuffer charBuffer,
                                                                                                                                                                 @NonNull ByteBuffer byteBuffer,
                                                                                                                                                                 boolean endOfInput,
                                                                                                                                                                 @NonNull ResponseStream output) throws IOException, InterruptedException, StreamingResponseCanceledException, CharacterCodingException {
                        CoderResult result;

                        do {
                                result = encoder.encode(charBuffer, byteBuffer, endOfInput);
                                writeEncodedBytesForSimulator(byteBuffer, output);

                                if (result.isError())
                                        result.throwException();
                        } while (result.isOverflow());
                }

                private void writeEncodedBytesForSimulator(@NonNull ByteBuffer byteBuffer,
                                                                                                                                                                                         @NonNull ResponseStream output) throws IOException, InterruptedException, StreamingResponseCanceledException {
                        byteBuffer.flip();
                        if (byteBuffer.hasRemaining())
                                output.write(byteBuffer);
                        byteBuffer.clear();
                }

                private void materializePublisher(com.soklet.StreamingResponseBody.@NonNull PublisherBody publisherBody,
                                @NonNull SimulatorResponseOutput output,
                                @NonNull SimulatorCancelationToken cancelationToken,
                                StreamLifecycleCoordinator.@NonNull Reservation reservation) throws Exception {
                        PublisherResponseStream.copy(publisherBody, cancelationToken, output, reservation,
                                        reservation::beginCleanup, throwable -> cancelSimulatorStream(cancelationToken, throwable),
                                        reservation::reportCleanupFailure);
                }

                @NonNull
                @Override
                public SseRequestResult performSseRequest(@NonNull Request request) {
                        Runnable releaseScope = enterScope(InternalLifecycleComponentType.SSE);
                        try (HttpTransportResponseFeedback feedback = HttpTransportResponseFeedback.create()) {
                                return performSseRequestWhileAdmitted(requireNonNull(request).copy().finish(), feedback);
                        } finally {
                                releaseScope.run();
                        }
                }

                @NonNull
                private SseRequestResult performSseRequestWhileAdmitted(
                                @NonNull Request request, @NonNull HttpTransportResponseFeedback feedback) {
                        MockSseServer sseServer = this.sseServer;

                        if (sseServer == null)
                                throw new IllegalStateException(format("You must specify a %s in your %s to simulate Server-Sent Event requests",
                                                SseServer.class.getSimpleName(), SokletConfig.class.getSimpleName()));

                        AtomicReference<@Nullable HttpRequestResult> requestResultHolder = new AtomicReference<>();
                        SseServer.RequestHandler requestHandler = sseServer.getRequestHandler().orElse(null);
                        AtomicReference<StreamLifecycleCoordinator.@Nullable Reservation> reservationHolder = new AtomicReference<>();
                        AtomicReference<SseConnection.@Nullable HandshakeFailureReason> replacementFailureReason = new AtomicReference<>();
                        AtomicReference<@Nullable Throwable> replacementFailureCause = new AtomicReference<>();

                        if (requestHandler == null)
                                throw new IllegalStateException("You must register a request handler prior to simulating SSE Event Source requests");
                        if (request.getHttpMethod() == HttpMethod.GET) {
                                try {
                                        DefaultSseServer.validateSseRequestBody(request);
                                } catch (IllegalRequestException failure) {
                                        return rejectSseRequestBeforeHandling(request, sseServer, failure);
                                }
                        }

                        requestHandler.handleRequest(request, HttpTransportResponseFeedback.fromWriter(feedback, requestResult -> {
                                HttpRequestResult effectiveResult = requestResult;
                                Throwable preparationFailure = null;
                                if (requestResult.getSseHandshakeResult().orElse(null) instanceof SseHandshakeResult.Accepted) {
                                        StreamLifecycleCoordinator.Reservation reservation = reserveSseStream();
                                        if (reservation == null) {
                                                replacementFailureReason.set(SseConnection.HandshakeFailureReason.CAPACITY_EXCEEDED);
                                                MarshaledResponse unavailable;
                                                try {
                                                        ResponseMarshaler marshaler = sseServer.getSokletConfig().orElseThrow().getResponseMarshaler();
                                                        unavailable = ResponseBodyCleanup.transform(
                                                                        marshaler.forServiceUnavailable(request, requestResult.getResourceMethod().orElse(null)),
                                                                        marshaler::postProcess);
                                                } catch (Throwable failure) {
                                                        preparationFailure = failure;
                                                        unavailable = DefaultSseServer.describeFailsafeHandshakeResponse(StatusCode.HTTP_500,
                                                                        request.getHttpMethod() == HttpMethod.HEAD);
                                                }
                                                effectiveResult = requestResult.copy().marshaledResponse(unavailable).response(null)
                                                                .sseHandshakeResult(null).finish();
                                        } else {
                                                reservationHolder.set(reservation);
                                        }
                                }
                                if (!(effectiveResult.getSseHandshakeResult().orElse(null) instanceof SseHandshakeResult.Accepted)
                                                && effectiveResult.getMarshaledResponse().isStreaming()) {
                                        preparationFailure = new IllegalArgumentException("SSE handshake rejection responses require a finite body");
                                        effectiveResult = effectiveResult.copy().sseHandshakeResult(null).response(null)
                                                        .marshaledResponse(DefaultSseServer.describeFailsafeHandshakeResponse(StatusCode.HTTP_500,
                                                                        request.getHttpMethod() == HttpMethod.HEAD)).finish();
                                }
                                if (preparationFailure != null) {
                                        replacementFailureReason.set(SseConnection.HandshakeFailureReason.INTERNAL_ERROR);
                                        replacementFailureCause.set(preparationFailure);
                                        sseServer.safelyLog(LogEvent.with(LogEventType.SSE_SERVER_INTERNAL_ERROR,
                                                        "Unable to generate SSE handshake response").throwable(preparationFailure).request(request)
                                                        .resourceMethod(requestResult.getResourceMethod().orElse(null)).build());
                                }
                                if (!sameInstance(effectiveResult, requestResult))
                                        effectiveResult = effectiveResult.copy().marshaledResponse(ResponseBodyCleanup.retain(
                                                        requestResult.getMarshaledResponse(), effectiveResult.getMarshaledResponse())).finish();
                                HttpRequestResult logicalResult = effectiveResult;
                                if (!(effectiveResult.getSseHandshakeResult().orElse(null) instanceof SseHandshakeResult.Accepted))
                                        effectiveResult = projectSimulatedSseRejection(request, effectiveResult);
                                requestResultHolder.set(effectiveResult);
                                if (!sameInstance(logicalResult, requestResult)) {
                                        return HttpTransportResponseOutcome.replaced(logicalResult.getMarshaledResponse(), preparationFailure, null);
                                }
                                return HttpTransportResponseOutcome.accepted();
                        }));

                        HttpRequestResult requestResult = requestResultHolder.get();
                        if (requestResult == null)
                                throw new IllegalStateException("SSE request handler did not provide a request result");

                        SseHandshakeResult sseHandshakeResult = requestResult.getSseHandshakeResult().orElse(null);
                        if (!(sseHandshakeResult instanceof SseHandshakeResult.Accepted))
                                requestResult = materializeFiniteSimulationResult(requestResult);

                        if (sseHandshakeResult == null) {
                                Throwable cause = replacementFailureCause.get();
                                if (cause == null)
                                        cause = requestResult.getRequestHandlingFailure().orElse(null);
                                SseConnection.HandshakeFailureReason reason = replacementFailureReason.get();
                                if (reason == null)
                                        reason = cause == null ? SseConnection.HandshakeFailureReason.HANDSHAKE_REJECTED
                                                        : SseConnection.HandshakeFailureReason.INTERNAL_ERROR;
                                notifySseHandshakeFailure(request, requestResult, reason, cause);
                                return new SseRequestResult.RequestFailed(requestResult);
                        }

                        if (sseHandshakeResult instanceof SseHandshakeResult.Accepted acceptedHandshake) {
                                SseClientInitializer clientInitializer = acceptedHandshake.getClientInitializer().orElse(null);
                                StreamLifecycleCoordinator.Reservation reservation = requireNonNull(reservationHolder.get());

                                try {
                                        requestResult = materializeAcceptedSseSimulationResult(requestResult);
                                        // Create a synthetic logical response using values from the accepted handshake.
                                        if (requestResult.getResponse().isEmpty())
                                                requestResult = requestResult.copy()
                                                                .response(Response.withStatusCode(200)
                                                                                .headers(acceptedHandshake.getHeaders())
                                                                                .cookies(acceptedHandshake.getCookies()).build()).finish();
                                        HandshakeAccepted handshakeAccepted = new HandshakeAccepted(acceptedHandshake, request,
                                                        requestResult, sseServer, reservation);
                                        if (!handshakeAccepted.initialize(clientInitializer))
                                                throw new IllegalStateException("The simulated SSE connection terminated before activation");
                                        return handshakeAccepted;
                                } catch (Throwable failure) {
                                        // An initializer already elected PRODUCER_FAILED; construction/activation
                                        // failures need the same reservation cleanup without replacing that winner.
                                        reservation.cancel(StreamTerminationReason.INTERNAL_ERROR, failure);
                                        reservation.complete();
                                        if (failure instanceof RuntimeException runtimeException)
                                                throw runtimeException;
                                        if (failure instanceof Error error)
                                                throw error;
                                        throw new IllegalStateException("The simulated SSE client initializer failed", failure);
                                }
                        }

                        if (sseHandshakeResult instanceof SseHandshakeResult.Rejected rejectedHandshake) {
                                notifySseHandshakeFailure(request, requestResult, SseConnection.HandshakeFailureReason.HANDSHAKE_REJECTED, null);
                                return new HandshakeRejected(rejectedHandshake, requestResult);
                        }

                        throw new IllegalStateException(format("Encountered unexpected %s: %s", SseHandshakeResult.class.getSimpleName(), sseHandshakeResult));
                }

                @NonNull
                private SseRequestResult rejectSseRequestBeforeHandling(@NonNull Request request,
                                @NonNull MockSseServer server, @NonNull IllegalRequestException failure) {
                        Object token = this.executionOwnerToken;
                        try (LifecycleExecutionContext.Scope ignored = token == null ? null : LifecycleExecutionContext.enter(token)) {
                                HttpRequestResult rejected = HttpRequestResult.withMarshaledResponse(MarshaledResponse.fromStatusCode(400))
                                                .requestHandlingFailure(failure).effectiveSseRequest(request).build();
                                // Live admission reports rejection even when its custom error marshaler fails.
                                notifySseHandshakeFailure(request, rejected, SseConnection.HandshakeFailureReason.HANDSHAKE_REJECTED, failure);
                                ResponseMarshaler marshaler = server.getSokletConfig().orElseThrow().getResponseMarshaler();
                                MarshaledResponse original = marshaler.forThrowable(request, failure, null);
                                MarshaledResponse response;
                                try {
                                        response = ResponseBodyCleanup.transform(original, marshaler::postProcess);
                                } catch (Throwable preparationFailure) {
                                        // The transform releases unchecked failures; JVM callbacks can also
                                        // emit checked failures that still need their owned channels closed.
                                        if (!(preparationFailure instanceof RuntimeException) && !(preparationFailure instanceof Error))
                                                MarshaledResponse.closeChannels(original.ownedChannels(), preparationFailure);
                                        throw preparationFailure;
                                }
                                boolean materializationOwnsCleanup = false;
                                try {
                                        enforceFinalResponseStatusCode(response.getStatusCode());
                                        if (response.isStreaming())
                                                throw new IllegalArgumentException("SSE handshake rejection responses require a finite body");
                                        HttpRequestResult result = projectSimulatedSseRejection(request,
                                                        rejected.copy().marshaledResponse(response).finish());
                                        // The materializer releases every owned channel, including when reading fails.
                                        materializationOwnsCleanup = true;
                                        return new SseRequestResult.RequestFailed(materializeFiniteSimulationResult(result));
                                } catch (Throwable preparationFailure) {
                                        if (!materializationOwnsCleanup)
                                                MarshaledResponse.closeChannels(response.ownedChannels(), preparationFailure);
                                        throw preparationFailure;
                                }
                        }
                }

                private void notifySseHandshakeFailure(@NonNull Request request, @NonNull HttpRequestResult requestResult,
                                SseConnection.@NonNull HandshakeFailureReason reason, @Nullable Throwable cause) {
                        MockSseServer server = requireNonNull(this.sseServer);
                        StreamLifecycleCoordinator coordinator;
                        synchronized (this.scopeStateLock) {
                                if (this.closed.get())
                                        return;
                                coordinator = sseCoordinatorWhileLocked();
                        }
                        Request effectiveRequest = requestResult.getEffectiveSseRequest().orElse(request);
                        ResourceMethod resourceMethod = requestResult.getResourceMethod().orElse(null);
                        if (!coordinator.dispatchRejectionObserver(() -> server.notifyConnectionLifecycle(effectiveRequest, resourceMethod,
                                        LogEventType.LIFECYCLE_OBSERVER_DID_ESTABLISH_SSE_CONNECTION_FAILED, "didFailToEstablishSseConnection",
                                        observer -> observer.didFailToEstablishSseConnection(effectiveRequest, resourceMethod, reason, cause),
                                        metrics -> metrics.didFailToEstablishSseConnection(effectiveRequest, resourceMethod, reason, cause)))) {
                                server.safelyLog(LogEvent.with(LogEventType.SSE_SERVER_CONNECTION_REJECTED,
                                                "Simulated SSE handshake-failure observer capacity was unavailable.").request(request).build());
                        }
                }

                @NonNull
                @Override
                public Simulator onBroadcastError(@Nullable Consumer<@NonNull Throwable> onBroadcastError) {
                        Runnable releaseScope = enterScope(InternalLifecycleComponentType.SSE);
                        try {
                                MockSseServer sseServer = this.sseServer;
                                if (sseServer != null)
                                        sseServer.onBroadcastError(onBroadcastError);
                                return this;
                        } finally {
                                releaseScope.run();
                        }
                }

                @NonNull
                @Override
                public Simulator onUnicastError(@Nullable Consumer<@NonNull Throwable> onUnicastError) {
                        Runnable releaseScope = enterScope(InternalLifecycleComponentType.SSE);
                        try {
                                MockSseServer sseServer = this.sseServer;
                                if (sseServer != null)
                                        sseServer.onUnicastError(onUnicastError);
                                return this;
                        } finally {
                                releaseScope.run();
                        }
                }

                @NonNull
                @Override
                public Optional<@NonNull HttpServer> getHttpServer() {
                        requireScopeOpen();
                        return Optional.ofNullable(this.server);
                }

                @NonNull
                @Override
                public Optional<@NonNull SseServer> getSseServer() {
                        requireScopeOpen();
                        return Optional.ofNullable(this.sseServer);
                }

                @NonNull
                @Override
                public Optional<@NonNull McpServer> getMcpServer() {
                        requireScopeOpen();
                        return Optional.ofNullable(this.mcpServer);
                }

                @NonNull
                private Runnable enterScope(
                                @NonNull InternalLifecycleComponentType kind) {
                        requireScopeOpen();
                        return this.scopeDispatchGate == null
                                        ? () -> {
                                        }
                                        : this.scopeDispatchGate.enter(requireNonNull(kind));
                }

                private void requireScopeOpen() {
                        requireScopeOpenWhileLocked();
                }

                private void requireScopeOpenWhileLocked() {
                        if (this.closed.get())
                                throw new IllegalStateException("The simulator scope is closed.");
                }

                @NonNull
                SimulatorOptions getSimulatorOptions() {
                        return this.simulatorOptions;
                }
        }

        @NotThreadSafe
        private static final class SimulatorResponseOutput implements ManagedResponseStream.Output {
                @NonNull
                private final ByteArrayOutputStream byteArrayOutputStream;
                @NonNull
                private final Integer limitInBytes;
                @NonNull
                private final SimulatorCancelationToken cancelationToken;
                private boolean closed;
                private final java.util.concurrent.atomic.@NonNull AtomicLong acceptedBytes;

                private SimulatorResponseOutput(@NonNull SimulatorCancelationToken cancelationToken,
                                @NonNull Integer limitInBytes, java.util.concurrent.atomic.@NonNull AtomicLong acceptedBytes) {
                        this.acceptedBytes = acceptedBytes;
                        this.byteArrayOutputStream = new ByteArrayOutputStream();
                        this.limitInBytes = requireNonNull(limitInBytes);
                        this.cancelationToken = requireNonNull(cancelationToken);
                }

                @Override
                public void write(@NonNull ByteBuffer byteBuffer) throws IOException, StreamingResponseCanceledException {
                        requireNonNull(byteBuffer);
                        this.cancelationToken.throwIfCanceled();

                        if (this.closed)
                                throw new StreamingResponseCanceledException(StreamTerminationReason.APPLICATION_CANCELED);

                        ByteBuffer source = byteBuffer.asReadOnlyBuffer();
                        int bytesToWrite = source.remaining();

                        if ((long) this.byteArrayOutputStream.size() + bytesToWrite > this.limitInBytes) {
                                this.cancelationToken.cancel(StreamTerminationReason.SIMULATOR_LIMIT_EXCEEDED, null);
                                throw new StreamingResponseCanceledException(StreamTerminationReason.SIMULATOR_LIMIT_EXCEEDED);
                        }

                        byte[] bytes = new byte[bytesToWrite];
                        source.get(bytes);
                        Runnable append = () -> {
                                this.byteArrayOutputStream.writeBytes(bytes);
                                this.acceptedBytes.addAndGet(bytesToWrite);
                                byteBuffer.position(byteBuffer.position() + bytesToWrite);
                        };
                        StreamLifecycleCoordinator.Reservation reservation = this.cancelationToken.reservation;
                        if (reservation == null) append.run();
                        else if (!reservation.acceptTransportOutput(append)) this.cancelationToken.throwIfCanceled();
                }

                @Override
                public int stagingCapacityInBytes() {
                        return Math.max(1, Math.min(8_192, this.limitInBytes));
                }

                @Override
                public void flush() throws StreamingResponseCanceledException {
                        this.cancelationToken.throwIfCanceled();
                }

                @Override
                public boolean isOpen() {
                        return !this.closed && !this.cancelationToken.isCanceled();
                }

                private byte @NonNull [] toByteArray() {
                        this.closed = true;
                        return this.byteArrayOutputStream.toByteArray();
                }
        }

        @ThreadSafe
        private static final class SimulatorCancelationToken implements CancelationToken {
                private static final @NonNull Runnable NO_CALLBACKS = () -> {};
                private final @NonNull Object lock = new Object();
                private boolean canceled;
                private boolean completed;
                private final StreamLifecycleCoordinator.@Nullable Reservation reservation;
                @Nullable
                private Set<@NonNull CancelationCallbackRegistration> callbacks;
                @NonNull
                private final Consumer<@NonNull Throwable> callbackFailureConsumer;
                @Nullable
                private StreamTerminationReason reason;
                @Nullable
                private Throwable cause;

                private SimulatorCancelationToken(@NonNull Consumer<@NonNull Throwable> callbackFailureConsumer,
                                StreamLifecycleCoordinator.@Nullable Reservation reservation) {
                        this.callbackFailureConsumer = requireNonNull(callbackFailureConsumer);
                        this.reservation = reservation;
                }

                @Override
                @NonNull
                public Boolean isCanceled() {
                        synchronized (this.lock) {
                                boolean canceled = this.canceled || this.reservation != null && this.reservation.isCanceled();
                                return !isCompleted() && canceled;
                        }
                }

                @Override
                @NonNull
                public Optional<@NonNull StreamTerminationReason> getCancelationReason() {
                        synchronized (this.lock) {
                                Optional<StreamTerminationReason> reason = this.reason == null && this.reservation != null
                                                ? this.reservation.reason() : Optional.ofNullable(this.reason);
                                return isCompleted() ? Optional.empty() : reason;
                        }
                }

                @Override
                @NonNull
                public Optional<@NonNull Throwable> getCancelationCause() {
                        synchronized (this.lock) {
                                Optional<Throwable> cause = this.reason == null && this.reservation != null
                                                ? this.reservation.cause() : Optional.ofNullable(this.cause);
                                return isCompleted() ? Optional.empty() : cause;
                        }
                }

                @Override
                @NonNull
                public CallbackRegistration onCancel(@NonNull Runnable callback) {
                        requireNonNull(callback);
                        CancelationCallbackRegistration registration = new CancelationCallbackRegistration(callback);
                        boolean runImmediately;

                        synchronized (this.lock) {
                                if (isCompleted()) {
                                        registration.callback = null;
                                        return registration;
                                }
                                runImmediately = this.canceled;
                                if (!runImmediately) {
                                        if (this.callbacks == null)
                                                this.callbacks = new LinkedHashSet<>();
                                        this.callbacks.add(registration);
                                }
                        }
                        if (runImmediately)
                                registration.invoke();
                        return registration;
                }

                private boolean cancel(@NonNull StreamTerminationReason reason,
                                @Nullable Throwable cause) {
                        requireNonNull(reason);
                        if (reason == StreamTerminationReason.COMPLETED)
                                throw new IllegalArgumentException("Cancelation reason cannot be COMPLETED");
                        if (this.reservation != null)
                                return this.reservation.cancel(reason, cause);
                        Runnable callbacks = reserveCancelation(reason, cause);
                        if (callbacks == null)
                                return false;
                        callbacks.run();
                        return true;
                }

                private void deliverCancelation(@NonNull StreamTerminationReason reason, @Nullable Throwable cause) {
                        Runnable callbacks = reserveCancelation(reason, cause);
                        if (callbacks != null && !sameInstance(callbacks, NO_CALLBACKS))
                                requireNonNull(this.reservation).dispatchCallbacks(callbacks);
                }

                @Nullable
                private Runnable reserveCancelation(@NonNull StreamTerminationReason reason, @Nullable Throwable cause) {

                        Set<CancelationCallbackRegistration> callbacksToRun;
                        synchronized (this.lock) {
                                if (this.canceled || isCompleted())
                                        return null;
                                this.reason = reason;
                                this.cause = cause;
                                this.canceled = true;
                                callbacksToRun = this.callbacks;
                                this.callbacks = null;
                        }
                        if (callbacksToRun == null || callbacksToRun.isEmpty())
                                return NO_CALLBACKS;
                        return () -> {
                                for (CancelationCallbackRegistration registration : callbacksToRun)
                                        registration.invoke();
                                callbacksToRun.clear();
                        };
                }

                private void complete() {
                        Set<CancelationCallbackRegistration> completedCallbacks;
                        synchronized (this.lock) {
                                if (this.canceled || this.completed || this.reservation != null
                                                && !this.reservation.isProductionComplete() && this.reservation.isCanceled())
                                        return;
                                this.completed = true;
                                completedCallbacks = this.callbacks;
                                this.callbacks = null;
                        }
                        // Completion may release many application registrations. Keep traversal
                        // outside the token monitor used by framework cancellation signals.
                        if (completedCallbacks != null) {
                                for (CancelationCallbackRegistration registration : completedCallbacks)
                                        registration.callback = null;
                                completedCallbacks.clear();
                        }
                }

                private boolean isCompleted() {
                        synchronized (this.lock) {
                                return this.completed || this.reservation != null && this.reservation.isProductionComplete();
                        }
                }

                private void runCallback(@NonNull Runnable callback) {
                        try {
                                callback.run();
                        } catch (Throwable throwable) {
                                try {
                                        this.callbackFailureConsumer.accept(throwable);
                                } catch (Throwable ignored) {
                                        // Diagnostic observers cannot suppress remaining callbacks.
                                }
                        }
                }

                private final class CancelationCallbackRegistration implements CallbackRegistration {
                        @Nullable
                        private volatile Runnable callback;

                        private CancelationCallbackRegistration(@NonNull Runnable callback) {
                                this.callback = callback;
                        }

                        @Override
                        public void close() {
                                synchronized (SimulatorCancelationToken.this.lock) {
                                        this.callback = null;
                                        if (SimulatorCancelationToken.this.callbacks != null)
                                                SimulatorCancelationToken.this.callbacks.remove(this);
                                }
                        }

                        private void invoke() {
                                Runnable callback;
                                synchronized (SimulatorCancelationToken.this.lock) {
                                        callback = this.callback;
                                        this.callback = null;
                                }
                                if (callback != null)
                                        runCallback(callback);
                        }
                }
        }

        /**
         * Mock server that doesn't touch the network at all, useful for testing.
         *
         * @author <a href="https://www.revetkn.com">Mark Allen</a>
         */
        @ThreadSafe
        static class MockHttpServer implements HttpServer {
                @NonNull
                private final TransportIdentity transportIdentity = TransportIdentity.create();
                private final int streamingLifecycleCapacity;
                private final int streamingCallbackConcurrency;
                @NonNull
                private final Duration streamingCleanupTimeout;
                @Nullable
                private volatile SokletConfig sokletConfig;
                private volatile HttpServer.@Nullable RequestHandler requestHandler;

                MockHttpServer() {
                        this(null);
                }

                MockHttpServer(@Nullable HttpServer sourceHttpServer) {
                        if (sourceHttpServer instanceof DefaultHttpServer defaultHttpServer) {
                                this.streamingLifecycleCapacity = defaultHttpServer.getStreamingLifecycleCapacity();
                                this.streamingCallbackConcurrency = defaultHttpServer.getStreamingCallbackConcurrency();
                                this.streamingCleanupTimeout = defaultHttpServer.getStreamingCleanupTimeout();
                        } else if (sourceHttpServer instanceof MockHttpServer mockHttpServer) {
                                this.streamingLifecycleCapacity = mockHttpServer.streamingLifecycleCapacity;
                                this.streamingCallbackConcurrency = mockHttpServer.streamingCallbackConcurrency;
                                this.streamingCleanupTimeout = mockHttpServer.streamingCleanupTimeout;
                        } else {
                                this.streamingLifecycleCapacity = DefaultHttpServer.DEFAULT_STREAMING_LIFECYCLE_CAPACITY;
                                this.streamingCallbackConcurrency = DefaultHttpServer.DEFAULT_STREAMING_CALLBACK_CONCURRENCY;
                                this.streamingCleanupTimeout = DefaultHttpServer.DEFAULT_STREAMING_CLEANUP_TIMEOUT;
                        }
                }

                @NonNull
                @Override
                public TransportIdentity getTransportIdentity() {
                        return this.transportIdentity;
                }

                @NonNull
                @Override
                public TransportRuntime attach(
                                @NonNull HttpTransportAttachmentContext attachmentContext,
                                @NonNull StartupContext startupContext) {
                        requireNonNull(startupContext);
                        HttpTransportAttachmentContext exactContext = requireNonNull(
                                        attachmentContext);
                        initialize(exactContext.getSokletConfig(),
                                        exactContext.getAdmissionFencedRequestHandler());
                        TransportTerminationSignal signal = exactContext.getTransportTerminationSignal();
                        return new TransportRuntime() {
                                @Override
                                public void start(@NonNull StartupContext context) {
                                        requireNonNull(context);
                                }

                                @Override
                                public void shutdownGracefully(@NonNull ShutdownContext context) {
                                        requireNonNull(context);
                                        signal.signalTerminated();
                                }

                                @Override
                                public void shutdownForcibly(@NonNull ShutdownContext context) {
                                        requireNonNull(context);
                                        signal.signalTerminated();
                                }
                        };
                }

                void start() {
                        // No-op
                }

                public void stop() {
                        // No-op
                }

                @NonNull
                public Boolean isStarted() {
                        return true;
                }

                public synchronized void initialize(@NonNull SokletConfig sokletConfig,
                                @NonNull RequestHandler requestHandler) {
                        requireNonNull(sokletConfig);
                        requireNonNull(requestHandler);

                        this.sokletConfig = sokletConfig;
                        this.requestHandler = requestHandler;
                }

                synchronized void releaseSimulationScopeState() {
                        this.sokletConfig = null;
                        this.requestHandler = null;
                }

                @NonNull
                synchronized Optional<@NonNull SokletConfig> getSokletConfig() {
                        return Optional.ofNullable(this.sokletConfig);
                }

                @NonNull
                synchronized Optional<@NonNull RequestHandler> getRequestHandler() {
                        return Optional.ofNullable(this.requestHandler);
                }
        }


        /**
         * Mock Server-Sent Event unicaster that doesn't touch the network at all, useful for testing.
         */
        @ThreadSafe
        static class MockSseUnicaster implements SseUnicaster {
                private final @NonNull Request request;
                private final @NonNull Consumer<@NonNull SseEvent> eventConsumer;
                private final @NonNull Consumer<@NonNull SseComment> commentConsumer;
                private final @NonNull Object initializerLock = new Object();
                private boolean initializing;

                MockSseUnicaster(@NonNull Request request,
                                @NonNull Consumer<@NonNull SseEvent> eventConsumer, @NonNull Consumer<@NonNull SseComment> commentConsumer) {
                        this.request = requireNonNull(request);
                        this.eventConsumer = requireNonNull(eventConsumer);
                        this.commentConsumer = requireNonNull(commentConsumer);
                }

                void beginInitializer() {
                        synchronized (this.initializerLock) { this.initializing = true; }
                }
                void finishInitializer() {
                        synchronized (this.initializerLock) { this.initializing = false; }
                }

                @Override public void unicastEvent(@NonNull SseEvent event) {
                        synchronized (this.initializerLock) {
                                if (!this.initializing)
                                        throw new IllegalStateException("The SSE unicaster is available only during client initialization");
                                this.eventConsumer.accept(requireNonNull(event));
                        }
                }
                @Override public void unicastComment(@NonNull SseComment comment) {
                        synchronized (this.initializerLock) {
                                if (!this.initializing)
                                        throw new IllegalStateException("The SSE unicaster is available only during client initialization");
                                this.commentConsumer.accept(requireNonNull(comment));
                        }
                }
                @Override public @NonNull ResourcePath getResourcePath() { return this.request.getResourcePath(); }
        }

        /**
         * Mock Server-Sent Event broadcaster that doesn't touch the network at all, useful for testing.
         */
        @ThreadSafe
        static class MockSseBroadcaster implements SseBroadcaster {
                // ConcurrentHashMap doesn't allow null values, so we use a sentinel if context is null
                private static final @NonNull Object NULL_CONTEXT_SENTINEL;

                static {
                        NULL_CONTEXT_SENTINEL = new Object();
                }

                @NonNull
                private final ResourcePath resourcePath;
                // Maps the Consumer (Listener) to its Context object (e.g. Locale)
                @NonNull
                private final Map<@NonNull Consumer<@NonNull SseEvent>, @NonNull Object> eventConsumers;
                // Same goes for comments
                @NonNull
                private final Map<@NonNull Consumer<@NonNull SseComment>, @NonNull Object> commentConsumers;
                private final @NonNull Set<@NonNull HandshakeAccepted> connections = ConcurrentHashMap.newKeySet();
                @NonNull
                private final AtomicReference<@Nullable Consumer<@NonNull Throwable>> broadcastErrorHandler;
                @NonNull
                private final Consumer<@NonNull LogEvent> logEventConsumer;

                public MockSseBroadcaster(@NonNull ResourcePath resourcePath,
                                                                                                                        @NonNull AtomicReference<@Nullable Consumer<@NonNull Throwable>> broadcastErrorHandler,
                                                                                                                        @NonNull Consumer<@NonNull LogEvent> logEventConsumer) {
                        requireNonNull(resourcePath);
                        requireNonNull(broadcastErrorHandler);
                        requireNonNull(logEventConsumer);

                        this.resourcePath = resourcePath;
                        this.eventConsumers = new ConcurrentHashMap<>();
                        this.commentConsumers = new ConcurrentHashMap<>();
                        this.broadcastErrorHandler = broadcastErrorHandler;
                        this.logEventConsumer = logEventConsumer;
                }

                @NonNull
                @Override
                public ResourcePath getResourcePath() {
                        return this.resourcePath;
                }

                @NonNull
                @Override
                public synchronized Long getClientCount() {
                        return Long.valueOf(getEventConsumers().size() + getCommentConsumers().size() - this.connections.size());
                }

                @Override
                public void broadcastEvent(@NonNull SseEvent sseEvent) {
                        requireNonNull(sseEvent);

                        for (Consumer<SseEvent> eventConsumer : getEventConsumers().keySet()) {
                                try {
                                        eventConsumer.accept(sseEvent);
                                } catch (Throwable throwable) {
                                        handleBroadcastError(throwable);
                                }
                        }
                }

                @Override
                public void broadcastComment(@NonNull SseComment sseComment) {
                        requireNonNull(sseComment);

                        for (Consumer<SseComment> commentConsumer : getCommentConsumers().keySet()) {
                                try {
                                        commentConsumer.accept(sseComment);
                                } catch (Throwable throwable) {
                                        handleBroadcastError(throwable);
                                }
                        }
                }

                @Override
                public <T> void broadcastEvent(@NonNull Function<@Nullable Object, @Nullable T> keySelector,
                                @NonNull Function<@Nullable T, @NonNull SseEvent> eventProvider) {
                        requireNonNull(keySelector);
                        requireNonNull(eventProvider);
                        Map<T, SseBroadcastPayload<SseEvent>> payloadCache = new HashMap<>();

                        getEventConsumers().forEach((consumer, context) -> {
                                try {
                                        Object clientContext = sameInstance(context, NULL_CONTEXT_SENTINEL) ? null : context;
                                        T key = keySelector.apply(clientContext);
                                        SseBroadcastPayload<SseEvent> generation = payloadCache.computeIfAbsent(key,
                                                        cacheKey -> SseBroadcastPayload.fromSupplier(() -> eventProvider.apply(cacheKey)));
                                        SseEvent event = generation.getPayload();
                                        if (event == null) {
                                                generation.recordFailedClient();
                                                return;
                                        }
                                        // A failing consumer cannot invalidate the payload shared by other clients.
                                        consumer.accept(event);
                                } catch (Throwable throwable) {
                                        handleBroadcastError(throwable);
                                }
                        });
                        payloadCache.values().forEach(generation -> handleBroadcastGenerationFailure(generation, false));
                }

                @Override
                public <T> void broadcastComment(@NonNull Function<@Nullable Object, @Nullable T> keySelector,
                                @NonNull Function<@Nullable T, @NonNull SseComment> commentProvider) {
                        requireNonNull(keySelector);
                        requireNonNull(commentProvider);
                        Map<T, SseBroadcastPayload<SseComment>> payloadCache = new HashMap<>();

                        getCommentConsumers().forEach((consumer, context) -> {
                                try {
                                        Object clientContext = sameInstance(context, NULL_CONTEXT_SENTINEL) ? null : context;
                                        T key = keySelector.apply(clientContext);
                                        SseBroadcastPayload<SseComment> generation = payloadCache.computeIfAbsent(key,
                                                        cacheKey -> SseBroadcastPayload.fromSupplier(() -> commentProvider.apply(cacheKey)));
                                        SseComment comment = generation.getPayload();
                                        if (comment == null) {
                                                generation.recordFailedClient();
                                                return;
                                        }
                                        consumer.accept(comment);
                                } catch (Throwable throwable) {
                                        handleBroadcastError(throwable);
                                }
                        });
                        payloadCache.values().forEach(generation -> handleBroadcastGenerationFailure(generation, true));
                }

                private void handleBroadcastGenerationFailure(@NonNull SseBroadcastPayload<?> generation, boolean comment) {
                        generation.getFailure().ifPresent(failure -> handleBroadcastError(failure,
                                        LogEventType.SSE_SERVER_BROADCAST_GENERATION_FAILED,
                                        format("Failed to generate %s for %d connections on %s",
                                                        comment ? "Server-Sent Event comment" : "Server-Sent Event",
                                                        generation.getFailedClientCount(), getResourcePath())));
                }

                @NonNull
                public Boolean registerEventConsumer(@NonNull Consumer<@NonNull SseEvent> eventConsumer) {
                        return registerEventConsumer(eventConsumer, null);
                }

                /**
                 * Registers a consumer with an associated context, simulating a client with specific traits.
                 */
                @NonNull
                public Boolean registerEventConsumer(@NonNull Consumer<@NonNull SseEvent> eventConsumer, @Nullable Object context) {
                        requireNonNull(eventConsumer);
                        // map.put returns null if the key was new, which conceptually matches "add" returning true
                        return this.getEventConsumers().put(eventConsumer, context == null ? NULL_CONTEXT_SENTINEL : context) == null;
                }

                @NonNull
                public Boolean unregisterEventConsumer(@NonNull Consumer<@NonNull SseEvent> eventConsumer) {
                        requireNonNull(eventConsumer);
                        return this.getEventConsumers().remove(eventConsumer) != null;
                }

                @NonNull
                public Boolean registerCommentConsumer(@NonNull Consumer<@NonNull SseComment> commentConsumer) {
                        return registerCommentConsumer(commentConsumer, null);
                }

                /**
                 * Registers a consumer with an associated context, simulating a client with specific traits.
                 */
                @NonNull
                public Boolean registerCommentConsumer(@NonNull Consumer<@NonNull SseComment> commentConsumer, @Nullable Object context) {
                        requireNonNull(commentConsumer);
                        return this.getCommentConsumers().put(commentConsumer, context == null ? NULL_CONTEXT_SENTINEL : context) == null;
                }

                @NonNull
                public Boolean unregisterCommentConsumer(@NonNull Consumer<@NonNull SseComment> commentConsumer) {
                        requireNonNull(commentConsumer);
                        return this.getCommentConsumers().remove(commentConsumer) != null;
                }

                @NonNull
                Map<@NonNull Consumer<@NonNull SseEvent>, @NonNull Object> getEventConsumers() {
                        return this.eventConsumers;
                }

                @NonNull
                Map<@NonNull Consumer<@NonNull SseComment>, @NonNull Object> getCommentConsumers() {
                        return this.commentConsumers;
                }

                synchronized void registerConnection(@NonNull HandshakeAccepted connection, @NonNull Consumer<@NonNull SseEvent> events,
                                @NonNull Consumer<@NonNull SseComment> comments, @Nullable Object context) {
                        this.connections.add(connection);
                        registerEventConsumer(events, context);
                        registerCommentConsumer(comments, context);
                }

                synchronized void unregisterConnection(@NonNull HandshakeAccepted connection, @NonNull Consumer<@NonNull SseEvent> events,
                                @NonNull Consumer<@NonNull SseComment> comments) {
                        unregisterEventConsumer(events);
                        unregisterCommentConsumer(comments);
                        this.connections.remove(connection);
                }

                synchronized void releaseSimulationScopeState() {
                        this.connections.clear();
                        this.eventConsumers.clear();
                        this.commentConsumers.clear();
                }

                void handleBroadcastError(@NonNull Throwable throwable) {
                        handleBroadcastError(throwable, LogEventType.SSE_SERVER_INTERNAL_ERROR,
                                        "SSE simulator broadcast consumer failed");
                }

                private void handleBroadcastError(@NonNull Throwable throwable,
                                @NonNull LogEventType logEventType, @NonNull String message) {
                        requireNonNull(throwable);
                        Consumer<Throwable> handler = this.broadcastErrorHandler.get();
                        if (handler != null) {
                                try {
                                        handler.accept(throwable);
                                        return;
                                } catch (Throwable ignored) {
                                        // Fall through to default behavior.
                                }
                        }
                        safelyLog(LogEvent.with(logEventType, message).throwable(throwable).build());
                }

                void safelyLog(@NonNull LogEvent logEvent) {
                        requireNonNull(logEvent);

                        try {
                                this.logEventConsumer.accept(logEvent);
                        } catch (Throwable ignored) {
                                // No safe fallback sink is available here.
                        }
                }
        }

        /**
         * Mock Server-Sent Event server that doesn't touch the network at all, useful for testing.
         *
         * @author <a href="https://www.revetkn.com">Mark Allen</a>
         */
        @ThreadSafe
        static class MockSseServer implements SseServer {
                @NonNull
                private final TransportIdentity transportIdentity;
                final int streamingLifecycleCapacity;
                final int connectionQueueCapacity;
                @Nullable
                private volatile SokletConfig sokletConfig;
                private volatile SseServer.@Nullable RequestHandler requestHandler;
                @NonNull
                private volatile Set<@NonNull ResourcePathDeclaration>
                                resourcePathDeclarations;
                private volatile boolean started;
                @NonNull
                private final ConcurrentHashMap<@NonNull ResourcePath, @NonNull MockSseBroadcaster> broadcastersByResourcePath;
                @NonNull
                private final AtomicReference<@Nullable Consumer<@NonNull Throwable>> broadcastErrorHandler;
                @NonNull
                private final AtomicReference<@Nullable Consumer<@NonNull Throwable>> unicastErrorHandler;

                public MockSseServer() { this(null); }

                MockSseServer(@Nullable SseServer sourceSseServer) {
                        if (sourceSseServer instanceof DefaultSseServer server) {
                                this.streamingLifecycleCapacity = server.getStreamingLifecycleCapacity();
                                this.connectionQueueCapacity = server.getConnectionQueueCapacity();
                        } else if (sourceSseServer instanceof MockSseServer server) {
                                this.streamingLifecycleCapacity = server.streamingLifecycleCapacity;
                                this.connectionQueueCapacity = server.connectionQueueCapacity;
                        } else {
                                this.streamingLifecycleCapacity = DefaultSseServer.DEFAULT_STREAMING_LIFECYCLE_CAPACITY;
                                this.connectionQueueCapacity = 128;
                        }
                        this.transportIdentity = TransportIdentity.create();
                        this.broadcastersByResourcePath = new ConcurrentHashMap<>();
                        this.broadcastErrorHandler = new AtomicReference<>();
                        this.unicastErrorHandler = new AtomicReference<>();
                        this.resourcePathDeclarations = Set.of();
                }

                @NonNull
                @Override
                public TransportIdentity getTransportIdentity() {
                        return this.transportIdentity;
                }

                @NonNull
                @Override
                public TransportRuntime attach(
                                @NonNull SseTransportAttachmentContext attachmentContext,
                                @NonNull StartupContext startupContext) {
                        requireNonNull(startupContext);
                        SseTransportAttachmentContext exactContext = requireNonNull(
                                        attachmentContext);
                        initialize(exactContext.getSokletConfig(),
                                        exactContext.getAdmissionFencedRequestHandler());
                        TransportTerminationSignal signal = exactContext.getTransportTerminationSignal();
                        return new TransportRuntime() {
                                @Override
                                public void start(@NonNull StartupContext context) {
                                        requireNonNull(context);
                                        MockSseServer.this.started = true;
                                }

                                @Override
                                public void shutdownGracefully(@NonNull ShutdownContext context) {
                                        requireNonNull(context);
                                        MockSseServer.this.started = false;
                                        signal.signalTerminated();
                                }

                                @Override
                                public void shutdownForcibly(@NonNull ShutdownContext context) {
                                        requireNonNull(context);
                                        MockSseServer.this.started = false;
                                        signal.signalTerminated();
                                }
                        };
                }

                void start() {
                        this.started = true;
                }

                public void stop() {
                        this.started = false;
                }

                @NonNull
                public Boolean isStarted() {
                        return this.started;
                }

                @NonNull
                @Override
                public synchronized Optional<? extends @NonNull SseBroadcaster> acquireBroadcaster(
                                @Nullable ResourcePath resourcePath) {
                        if (resourcePath == null || !this.started
                                        || this.sokletConfig == null)
                                return Optional.empty();
                        if (this.resourcePathDeclarations.stream().noneMatch(
                                        declaration -> declaration.matches(resourcePath)))
                                return Optional.empty();

                        MockSseBroadcaster broadcaster = getBroadcastersByResourcePath()
                                        .computeIfAbsent(resourcePath, rp -> new MockSseBroadcaster(rp, broadcastErrorHandler, this::safelyLog));

                        return Optional.of(broadcaster);
                }

                void registerConnection(@NonNull HandshakeAccepted connection, @NonNull ResourcePath resourcePath,
                                @NonNull Consumer<@NonNull SseEvent> events, @NonNull Consumer<@NonNull SseComment> comments, @Nullable Object context) {
                        this.broadcastersByResourcePath.computeIfAbsent(resourcePath,
                                        rp -> new MockSseBroadcaster(rp, this.broadcastErrorHandler, this::safelyLog))
                                        .registerConnection(connection, events, comments, context);
                }

                void unregisterConnection(@NonNull HandshakeAccepted connection, @NonNull ResourcePath resourcePath,
                                @NonNull Consumer<@NonNull SseEvent> events, @NonNull Consumer<@NonNull SseComment> comments) {
                        MockSseBroadcaster broadcaster = this.broadcastersByResourcePath.get(resourcePath);
                        if (broadcaster != null)
                                broadcaster.unregisterConnection(connection, events, comments);
                }

                public void registerEventConsumer(@NonNull ResourcePath resourcePath,
                                                                                                                                                        @NonNull Consumer<@NonNull SseEvent> eventConsumer) {
                        registerEventConsumer(resourcePath, eventConsumer, null);
                }

                public void registerEventConsumer(@NonNull ResourcePath resourcePath,
                                                                                                                                                        @NonNull Consumer<@NonNull SseEvent> eventConsumer,
                                                                                                                                                        @Nullable Object context) {
                        requireNonNull(resourcePath);
                        requireNonNull(eventConsumer);

                        MockSseBroadcaster broadcaster = getBroadcastersByResourcePath()
                                        .computeIfAbsent(resourcePath, rp -> new MockSseBroadcaster(rp, broadcastErrorHandler, this::safelyLog));

                        broadcaster.registerEventConsumer(eventConsumer, context);
                }

                @NonNull
                public Boolean unregisterEventConsumer(@NonNull ResourcePath resourcePath,
                                                                                                                                                                         @NonNull Consumer<@NonNull SseEvent> eventConsumer) {
                        requireNonNull(resourcePath);
                        requireNonNull(eventConsumer);

                        MockSseBroadcaster broadcaster = getBroadcastersByResourcePath().get(resourcePath);

                        if (broadcaster == null)
                                return false;

                        return broadcaster.unregisterEventConsumer(eventConsumer);
                }

                public void registerCommentConsumer(@NonNull ResourcePath resourcePath,
                                                                                                                                                                @NonNull Consumer<@NonNull SseComment> commentConsumer) {
                        registerCommentConsumer(resourcePath, commentConsumer, null);
                }

                public void registerCommentConsumer(@NonNull ResourcePath resourcePath,
                                                                                                                                                                @NonNull Consumer<@NonNull SseComment> commentConsumer,
                                                                                                                                                                @Nullable Object context) {
                        requireNonNull(resourcePath);
                        requireNonNull(commentConsumer);

                        MockSseBroadcaster broadcaster = getBroadcastersByResourcePath()
                                        .computeIfAbsent(resourcePath, rp -> new MockSseBroadcaster(rp, broadcastErrorHandler, this::safelyLog));

                        broadcaster.registerCommentConsumer(commentConsumer, context);
                }

                @NonNull
                public Boolean unregisterCommentConsumer(@NonNull ResourcePath resourcePath,
                                                                                                                                                                                 @NonNull Consumer<@NonNull SseComment> commentConsumer) {
                        requireNonNull(resourcePath);
                        requireNonNull(commentConsumer);

                        MockSseBroadcaster broadcaster = getBroadcastersByResourcePath().get(resourcePath);

                        if (broadcaster == null)
                                return false;

                        return broadcaster.unregisterCommentConsumer(commentConsumer);
                }

                public synchronized void initialize(@NonNull SokletConfig sokletConfig,
                                SseServer.@NonNull RequestHandler requestHandler) {
                        requireNonNull(sokletConfig);
                        requireNonNull(requestHandler);

                        this.sokletConfig = sokletConfig;
                        this.requestHandler = requestHandler;
                        this.resourcePathDeclarations = Set.copyOf(sokletConfig
                                        .getResourceMethodResolver().getResourceMethods().stream()
                                        .filter(ResourceMethod::isSseEventSource)
                                        .map(ResourceMethod::getResourcePathDeclaration)
                                        .toList());
                }

                synchronized void releaseSimulationScopeState() {
                        this.started = false;
                        this.sokletConfig = null;
                        this.requestHandler = null;
                        this.resourcePathDeclarations = Set.of();
                        for (MockSseBroadcaster broadcaster
                                        : this.broadcastersByResourcePath.values())
                                broadcaster.releaseSimulationScopeState();
                        this.broadcastersByResourcePath.clear();
                        this.broadcastErrorHandler.set(null);
                        this.unicastErrorHandler.set(null);
                }

                public void onBroadcastError(@Nullable Consumer<@NonNull Throwable> onBroadcastError) {
                        this.broadcastErrorHandler.set(onBroadcastError);
                }

                public void onUnicastError(@Nullable Consumer<@NonNull Throwable> onUnicastError) {
                        this.unicastErrorHandler.set(onUnicastError);
                }

                void notifyConnectionLifecycle(@NonNull Request request, @Nullable ResourceMethod resourceMethod,
                                @NonNull LogEventType failureType, @NonNull String callbackName,
                                @NonNull Consumer<@NonNull LifecycleObserver> observerNotification,
                                @NonNull Consumer<@NonNull MetricsCollector> metricsNotification) {
                        SokletConfig config = this.sokletConfig;
                        if (config == null)
                                return;
                        try {
                                observerNotification.accept(config.getAggregateLifecycleObserver());
                        } catch (Throwable failure) {
                                safelyLog(LogEvent.with(failureType, "An exception occurred while invoking LifecycleObserver::" + callbackName)
                                                .request(request).resourceMethod(resourceMethod).throwable(failure).build());
                        }
                        try {
                                metricsNotification.accept(config.getMetricsCollector());
                        } catch (Throwable failure) {
                                safelyLog(LogEvent.with(LogEventType.METRICS_COLLECTOR_FAILED,
                                                "An exception occurred while invoking MetricsCollector::" + callbackName)
                                                .request(request).resourceMethod(resourceMethod).throwable(failure).build());
                        }
                }

                void safelyLog(@NonNull LogEvent logEvent) {
                        requireNonNull(logEvent);

                        SokletConfig sokletConfig = this.sokletConfig;

                        if (sokletConfig == null)
                                return;

                        try {
                                sokletConfig.getAggregateLifecycleObserver().didReceiveLogEvent(logEvent);
                        } catch (Throwable observerFailure) {
                                LifecycleObserverLogFallback.report(observerFailure);
                        }
                }

                @NonNull
                synchronized Optional<@NonNull SokletConfig> getSokletConfig() {
                        return Optional.ofNullable(this.sokletConfig);
                }

                @NonNull
                synchronized Optional<SseServer.@NonNull RequestHandler> getRequestHandler() {
                        return Optional.ofNullable(this.requestHandler);
                }

                @NonNull
                ConcurrentHashMap<@NonNull ResourcePath, @NonNull MockSseBroadcaster> getBroadcastersByResourcePath() {
                        return this.broadcastersByResourcePath;
                }

                @NonNull
                AtomicReference<@Nullable Consumer<@NonNull Throwable>> getBroadcastErrorHandler() {
                        return this.broadcastErrorHandler;
                }

                @NonNull
                AtomicReference<@Nullable Consumer<@NonNull Throwable>> getUnicastErrorHandler() {
                        return this.unicastErrorHandler;
                }
        }

}