diff --git a/src/FreeDSx/Ldap/Container/HandlerContainerProvider.php b/src/FreeDSx/Ldap/Container/HandlerContainerProvider.php
index 54449463..0669e6cf 100644
--- a/src/FreeDSx/Ldap/Container/HandlerContainerProvider.php
+++ b/src/FreeDSx/Ldap/Container/HandlerContainerProvider.php
@@ -82,7 +82,7 @@ private function makeFactoryMap(Container $container): ProtocolHandlerFactoryMap
             HandlerId::StartTls->value => static fn(HandlerContext $context): ServerProtocolHandlerInterface
                 => new ServerStartTlsHandler(
                     options: $container->get(ServerOptions::class),
-                    connection: $context->queue,
+                    connection: $context->connection,
                     eventLogger: $context->eventLogger,
                 ),
             HandlerId::UnsupportedExtended->value => static fn(): ServerProtocolHandlerInterface
@@ -178,7 +178,6 @@ private function makeSyncHandler(
         if ($journal !== null) {
             $stream = new ChangeStream($journal);
             $streamer = new SyncPersistStreamer(
-                queue: $context->queue,
                 backend: $backend,
                 projector: $projector,
                 stream: $stream,
@@ -191,7 +190,6 @@ private function makeSyncHandler(
         }
 
         return new ServerSyncHandler(
-            queue: $context->queue,
             backend: $backend,
             projector: $projector,
             limits: $searchLimits ?? $options->makeSearchLimits(),
diff --git a/src/FreeDSx/Ldap/Protocol/Factory/HandlerContext.php b/src/FreeDSx/Ldap/Protocol/Factory/HandlerContext.php
index d20a4e81..645e3901 100644
--- a/src/FreeDSx/Ldap/Protocol/Factory/HandlerContext.php
+++ b/src/FreeDSx/Ldap/Protocol/Factory/HandlerContext.php
@@ -13,7 +13,7 @@
 
 namespace FreeDSx\Ldap\Protocol\Factory;
 
-use FreeDSx\Ldap\Protocol\Queue\ServerQueue;
+use FreeDSx\Ldap\Protocol\Queue\ConnectionControl;
 use FreeDSx\Ldap\Server\Logging\EventLogger;
 use FreeDSx\Ldap\Server\PasswordPolicy\PasswordPolicyContext;
 use FreeDSx\Ldap\Server\RequestHistory;
@@ -27,7 +27,7 @@
 final readonly class HandlerContext
 {
     public function __construct(
-        public ServerQueue $queue,
+        public ConnectionControl $connection,
         public EventLogger $eventLogger,
         public RequestHistory $requestHistory,
         public ?PasswordPolicyContext $passwordPolicyContext = null,
diff --git a/src/FreeDSx/Ldap/Protocol/Queue/Response/ResponseStream.php b/src/FreeDSx/Ldap/Protocol/Queue/Response/ResponseStream.php
index 61425b73..e3313ef7 100644
--- a/src/FreeDSx/Ldap/Protocol/Queue/Response/ResponseStream.php
+++ b/src/FreeDSx/Ldap/Protocol/Queue/Response/ResponseStream.php
@@ -146,4 +146,12 @@ public static function none(OperationResult $outcome): self
             $outcome,
         );
     }
+
+    /**
+     * An already-written response carrying only its resolved outcome, handed up past the writer.
+     */
+    public static function resolved(OperationResult $outcome): self
+    {
+        return self::none($outcome);
+    }
 }
diff --git a/src/FreeDSx/Ldap/Protocol/ServerProtocolHandler/ServerSyncHandler.php b/src/FreeDSx/Ldap/Protocol/ServerProtocolHandler/ServerSyncHandler.php
index d562e1ae..e3dac84b 100644
--- a/src/FreeDSx/Ldap/Protocol/ServerProtocolHandler/ServerSyncHandler.php
+++ b/src/FreeDSx/Ldap/Protocol/ServerProtocolHandler/ServerSyncHandler.php
@@ -26,8 +26,9 @@
 use FreeDSx\Ldap\Operation\ResultCode;
 use FreeDSx\Ldap\Protocol\LdapMessageRequest;
 use FreeDSx\Ldap\Protocol\LdapMessageResponse;
+use FreeDSx\Ldap\Protocol\Queue\Response\Cancellation;
+use FreeDSx\Ldap\Protocol\Queue\Response\QueueWriterConfig;
 use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
-use FreeDSx\Ldap\Protocol\Queue\ServerQueue;
 use FreeDSx\Ldap\Server\Backend\LdapBackendInterface;
 use FreeDSx\Ldap\Server\Backend\Storage\Journal\Read\ChangeStream;
 use FreeDSx\Ldap\Server\Operation\SearchOperationResult;
@@ -50,7 +51,6 @@ final class ServerSyncHandler implements ServerProtocolHandlerInterface
     use ServerSearchTrait;
 
     public function __construct(
-        private readonly ServerQueue $queue,
         private readonly LdapBackendInterface $backend,
         private readonly SyncResultProjector $projector,
         private readonly SearchLimits $limits = new SearchLimits(),
@@ -60,8 +60,6 @@ public function __construct(
     ) {}
 
     /**
-     * Phase 1: sync (refresh + persist) stays a direct writer; it is folded into the writer in Phase 2.
-     *
      * @throws OperationException
      */
     public function handleRequest(
@@ -84,6 +82,7 @@ public function handleRequest(
         $this->assertModeSupported($mode);
 
         $baseDn = $this->assertBaseDnProvided($request);
+        $messageId = $message->getMessageId();
         $latestSeq = $stream->latestSeq();
         $sinceSeq = $this->resolveSince($control, $stream, $latestSeq);
         $state = new SearchResultState();
@@ -93,37 +92,62 @@ public function handleRequest(
             : $this->incrementalEntries($message, $request, $token, $streamer, $sinceSeq, $state);
 
         $cookie = (new SyncCookie($stream->origin(), $latestSeq))->encode();
+        $outcome = fn(): SearchOperationResult => SearchOperationResult::success(
+            $message,
+            $state->entriesReturned,
+        );
 
         if ($mode === SyncRequestControl::MODE_REFRESH_AND_PERSIST) {
-            $this->queue->sendMessages($this->withRefreshDone(
-                $entries,
-                $message->getMessageId(),
-                $sinceSeq === null
-                    ? new SyncRefreshPresent(true, $cookie)
-                    : new SyncRefreshDelete(true, $cookie),
-            ));
-            $streamer->persist(
-                $latestSeq,
-                $request,
-                $token,
-                $message->getMessageId(),
-                $baseDn,
+            $cancellation = new Cancellation();
+            $boundary = $sinceSeq === null
+                ? new SyncRefreshPresent(true, $cookie)
+                : new SyncRefreshDelete(true, $cookie);
+
+            return ResponseStream::streaming(
+                $this->concat(
+                    $this->withRefreshDone(
+                        $entries,
+                        $messageId,
+                        $boundary,
+                    ),
+                    $streamer->stream(
+                        $latestSeq,
+                        $request,
+                        $token,
+                        $messageId,
+                        $baseDn,
+                        $cancellation,
+                    ),
+                ),
+                $outcome,
+                // Each change and keepalive flushes immediately for liveness; poll cancel after each.
+                new QueueWriterConfig(flushPerMessage: true, signalInterval: 1),
+                $cancellation,
             );
-        } else {
-            // refreshDeletes: a delete phase (true) only for an incremental sync that sent explicit
-            // deletes; a full refresh is a present phase (false) the consumer reconciles by absence.
-            $this->queue->sendMessages($this->withSyncDone(
+        }
+
+        // refreshDeletes: a delete phase (true) only for an incremental sync that sent explicit
+        // deletes; a full refresh is a present phase (false) the consumer reconciles by absence.
+        return ResponseStream::streaming(
+            $this->withSyncDone(
                 $entries,
-                $message->getMessageId(),
+                $messageId,
                 $baseDn,
                 new SyncDoneControl($cookie, $sinceSeq !== null),
-            ));
-        }
+            ),
+            $outcome,
+        );
+    }
 
-        return ResponseStream::none(SearchOperationResult::success(
-            $message,
-            $state->entriesReturned,
-        ));
+    /**
+     * @param Generator<LdapMessageResponse> ...$streams
+     * @return Generator<LdapMessageResponse>
+     */
+    private function concat(Generator ...$streams): Generator
+    {
+        foreach ($streams as $stream) {
+            yield from $stream;
+        }
     }
 
     /**
diff --git a/src/FreeDSx/Ldap/Server/ConnectionHandlerBuilder.php b/src/FreeDSx/Ldap/Server/ConnectionHandlerBuilder.php
index d868a957..1fc22778 100644
--- a/src/FreeDSx/Ldap/Server/ConnectionHandlerBuilder.php
+++ b/src/FreeDSx/Ldap/Server/ConnectionHandlerBuilder.php
@@ -56,7 +56,7 @@
 use FreeDSx\Ldap\Server\Middleware\MetricsMiddleware;
 use FreeDSx\Ldap\Server\Middleware\OperationAuditMiddleware;
 use FreeDSx\Ldap\Server\Middleware\OperationAuthorizationMiddleware;
-use FreeDSx\Ldap\Server\Middleware\OperationErrorMiddleware;
+use FreeDSx\Ldap\Server\Middleware\ResponseWriterMiddleware;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\HandlerInvoker;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareChain;
 use FreeDSx\Ldap\Server\Middleware\ReadOnlyMiddleware;
@@ -352,7 +352,7 @@ private function makeProtocolHandlerProvider(
             routeResolver: $this->container->get(ServerProtocolHandlerFactory::class),
             factories: $this->container->get(ProtocolHandlerFactoryMap::class),
             context: new HandlerContext(
-                queue: $queue,
+                connection: $queue,
                 eventLogger: $eventLogger,
                 requestHistory: $requestHistory,
                 passwordPolicyContext: $policyContext,
@@ -393,28 +393,24 @@ private function makeRequestPipeline(
                 ),
                 // The token is resolved at this point, so per-identity limits can be attached.
                 $this->container->get(ResourceLimitMiddleware::class),
-                new OperationErrorMiddleware(
-                    $queue,
+                // Audits the resolved outcome; sits above the writer so a streamed result's count is final.
+                new OperationAuditMiddleware(new OperationAuditor($eventLogger)),
+                // The single sink: drains every operation response and renders any thrown failure.
+                new ResponseWriterMiddleware(
+                    new ResponseWriter($queue),
                     $backend,
                     $options->getAccessControl(),
                 ),
-                new OperationAuditMiddleware(new OperationAuditor($eventLogger)),
-                // Only present on a read-only replica; sits below OperationErrorMiddleware so a rejection renders,
+                // Only present on a read-only replica; sits below the writer so its referral is drained,
                 // and before the ACL loop so a write short-circuits early.
                 ...($replicaConfig !== null
-                    ? [new ReadOnlyMiddleware(
-                        $queue,
-                        $replicaConfig,
-                    )]
+                    ? [new ReadOnlyMiddleware($replicaConfig)]
                     : []),
                 $this->container->get(CriticalControlMiddleware::class),
                 $this->container->get(OperationAuthorizationMiddleware::class),
                 $this->container->get(AssertionMiddleware::class),
             ],
-            new HandlerInvoker(
-                $handlerProvider,
-                new ResponseWriter($queue),
-            ),
+            new HandlerInvoker($handlerProvider),
         );
     }
 }
diff --git a/src/FreeDSx/Ldap/Server/Middleware/AssertionMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/AssertionMiddleware.php
index 43abfd38..c08ee1ef 100644
--- a/src/FreeDSx/Ldap/Server/Middleware/AssertionMiddleware.php
+++ b/src/FreeDSx/Ldap/Server/Middleware/AssertionMiddleware.php
@@ -17,12 +17,12 @@
 use FreeDSx\Ldap\Control\PagingControl;
 use FreeDSx\Ldap\Exception\OperationException;
 use FreeDSx\Ldap\Operation\Request\SearchRequest;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Protocol\ServerProtocolHandler\AssertionEvaluator;
 use FreeDSx\Ldap\Server\AccessControl\OperationTargetDn;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
 
 /**
  * Rejects an operation whose RFC 4528 assertion control does not match its target entry, before dispatch.
@@ -40,7 +40,7 @@ public function __construct(private AssertionEvaluator $evaluator) {}
     public function process(
         ServerRequestContext $context,
         MiddlewareHandlerInterface $next,
-    ): OperationResult {
+    ): ResponseStream {
         $message = $context->message;
         $request = $message->getRequest();
 
diff --git a/src/FreeDSx/Ldap/Server/Middleware/AuthorizationResolutionMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/AuthorizationResolutionMiddleware.php
index e04fe9af..1ca3e777 100644
--- a/src/FreeDSx/Ldap/Server/Middleware/AuthorizationResolutionMiddleware.php
+++ b/src/FreeDSx/Ldap/Server/Middleware/AuthorizationResolutionMiddleware.php
@@ -17,10 +17,10 @@
 use FreeDSx\Ldap\Exception\OperationException;
 use FreeDSx\Ldap\Operation\ResultCode;
 use FreeDSx\Ldap\Protocol\Authorization\DispatchAuthorizer;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
 use FreeDSx\Ldap\Server\PasswordPolicy\Decision\PasswordPolicyOutcome;
 use FreeDSx\Ldap\Server\PasswordPolicy\PasswordPolicyContext;
 
@@ -45,7 +45,7 @@ public function __construct(
     public function process(
         ServerRequestContext $context,
         MiddlewareHandlerInterface $next,
-    ): OperationResult {
+    ): ResponseStream {
         $authorization = $this->dispatchAuthorizer->authorize($context->message);
 
         if ($authorization->requiresAuthentication()) {
diff --git a/src/FreeDSx/Ldap/Server/Middleware/BindMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/BindMiddleware.php
index 6e8a9e8a..4434b754 100644
--- a/src/FreeDSx/Ldap/Server/Middleware/BindMiddleware.php
+++ b/src/FreeDSx/Ldap/Server/Middleware/BindMiddleware.php
@@ -16,12 +16,12 @@
 use FreeDSx\Ldap\Exception\OperationException;
 use FreeDSx\Ldap\Operation\ResultCode;
 use FreeDSx\Ldap\Protocol\Authenticator;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Protocol\ServerAuthorization;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
 use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
 use FreeDSx\Ldap\Server\Token\AnonToken;
 
 /**
@@ -43,7 +43,7 @@ public function __construct(
     public function process(
         ServerRequestContext $context,
         MiddlewareHandlerInterface $next,
-    ): OperationResult {
+    ): ResponseStream {
         $request = $context->message->getRequest();
 
         if (!$this->authorization->isAuthenticationRequest($request)) {
@@ -60,8 +60,9 @@ public function process(
             );
         }
 
+        // The authenticator writes the bind response itself (an interactive sub-protocol); carry only the outcome.
         $this->authorization->setToken($this->authenticator->bind($context->message));
 
-        return OperationOutcomeResult::succeeded();
+        return ResponseStream::resolved(OperationOutcomeResult::succeeded());
     }
 }
diff --git a/src/FreeDSx/Ldap/Server/Middleware/CriticalControlMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/CriticalControlMiddleware.php
index 91a8e3af..da4027c7 100644
--- a/src/FreeDSx/Ldap/Server/Middleware/CriticalControlMiddleware.php
+++ b/src/FreeDSx/Ldap/Server/Middleware/CriticalControlMiddleware.php
@@ -17,10 +17,10 @@
 use FreeDSx\Ldap\Exception\OperationException;
 use FreeDSx\Ldap\Operation\ResultCode;
 use FreeDSx\Ldap\Protocol\Factory\HandlerRouteResolverInterface;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
 
 use function in_array;
 use function sprintf;
@@ -44,7 +44,7 @@ public function __construct(
     public function process(
         ServerRequestContext $context,
         MiddlewareHandlerInterface $next,
-    ): OperationResult {
+    ): ResponseStream {
         $controls = $context->message->controls();
         $routeId = $this->routeResolver->routeIdFor(
             $context->message->getRequest(),
diff --git a/src/FreeDSx/Ldap/Server/Middleware/MetricsMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/MetricsMiddleware.php
index 6b7716f3..a91a5591 100644
--- a/src/FreeDSx/Ldap/Server/Middleware/MetricsMiddleware.php
+++ b/src/FreeDSx/Ldap/Server/Middleware/MetricsMiddleware.php
@@ -21,6 +21,7 @@
 use FreeDSx\Ldap\Operation\Request\SearchRequest;
 use FreeDSx\Ldap\Operation\Request\SimpleBindRequest;
 use FreeDSx\Ldap\Operation\ResultCode;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Server\Metrics\MetricsRecorderInterface;
 use FreeDSx\Ldap\Server\Metrics\Observation\OperationObservation;
 use FreeDSx\Ldap\Server\Metrics\Rollup\OperationRollupCoordinator;
@@ -28,7 +29,6 @@
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
 use FreeDSx\Ldap\Server\Operation\OperationOutcome;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
 use Throwable;
 
 use function microtime;
@@ -53,15 +53,16 @@ public function __construct(
     public function process(
         ServerRequestContext $context,
         MiddlewareHandlerInterface $next,
-    ): OperationResult {
+    ): ResponseStream {
         $request = $context->message->getRequest();
         $operation = OperationType::classify($request);
         $this->recorder->operationStarted($operation);
         $startedAt = microtime(true);
 
         try {
-            $result = $next->handle($context);
+            $stream = $next->handle($context);
         } catch (OperationException $e) {
+            // Only front-of-chain failures (bind/authorization) reach here; operation failures resolve as outcomes.
             $this->record(
                 $request,
                 $operation,
@@ -84,6 +85,7 @@ public function process(
             throw $e;
         }
 
+        $result = $stream->outcome();
         $this->record(
             $request,
             $operation,
@@ -92,7 +94,7 @@ public function process(
             $result->resultCode(),
         );
 
-        return $result;
+        return $stream;
     }
 
     private function record(
diff --git a/src/FreeDSx/Ldap/Server/Middleware/OperationAuditMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/OperationAuditMiddleware.php
index 2998010f..0bcd065e 100644
--- a/src/FreeDSx/Ldap/Server/Middleware/OperationAuditMiddleware.php
+++ b/src/FreeDSx/Ldap/Server/Middleware/OperationAuditMiddleware.php
@@ -13,19 +13,15 @@
 
 namespace FreeDSx\Ldap\Server\Middleware;
 
-use FreeDSx\Ldap\Exception\OperationException;
-use FreeDSx\Ldap\Exception\SchemaRuleException;
-use FreeDSx\Ldap\Operation\Request\CompareRequest;
-use FreeDSx\Ldap\Operation\Request\SearchRequest;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Server\Logging\OperationAuditor;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
 use FreeDSx\Ldap\Server\Operation\AuditableResult;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
 
 /**
- * Audits every operation outcome (the success result, or a thrown failure) the pipeline propagates back up.
+ * Audits every operation outcome the pipeline resolves (success or a writer-answered failure).
  *
  * @internal
  * @author Chad Sikorra <Chad.Sikorra@gmail.com>
@@ -34,23 +30,12 @@
 {
     public function __construct(private OperationAuditor $auditor) {}
 
-    /**
-     * @throws OperationException
-     */
     public function process(
         ServerRequestContext $context,
         MiddlewareHandlerInterface $next,
-    ): OperationResult {
-        try {
-            $result = $next->handle($context);
-        } catch (OperationException $e) {
-            $this->recordFailure(
-                $context,
-                $e,
-            );
-
-            throw $e;
-        }
+    ): ResponseStream {
+        $stream = $next->handle($context);
+        $result = $stream->outcome();
 
         if ($result instanceof AuditableResult) {
             $result->record(
@@ -59,45 +44,6 @@ public function process(
             );
         }
 
-        return $result;
-    }
-
-    private function recordFailure(
-        ServerRequestContext $context,
-        OperationException $e,
-    ): void {
-        if ($e instanceof SchemaRuleException) {
-            $this->auditor->recordSchemaViolations(
-                $e->getViolations(),
-                $context->message,
-                $context->tokenOrFail(),
-            );
-        }
-
-        if ($context->message->getRequest() instanceof SearchRequest) {
-            $this->auditor->recordSearchFailure(
-                $context->message,
-                $e,
-                $context->tokenOrFail(),
-            );
-
-            return;
-        }
-
-        if ($context->message->getRequest() instanceof CompareRequest) {
-            $this->auditor->recordCompareFailure(
-                $context->message,
-                $e,
-                $context->tokenOrFail(),
-            );
-
-            return;
-        }
-
-        $this->auditor->recordFailure(
-            $context->message,
-            $e,
-            $context->tokenOrFail(),
-        );
+        return $stream;
     }
 }
diff --git a/src/FreeDSx/Ldap/Server/Middleware/OperationAuthorizationMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/OperationAuthorizationMiddleware.php
index 2ff8fb5c..0814aa1a 100644
--- a/src/FreeDSx/Ldap/Server/Middleware/OperationAuthorizationMiddleware.php
+++ b/src/FreeDSx/Ldap/Server/Middleware/OperationAuthorizationMiddleware.php
@@ -28,6 +28,7 @@
 use FreeDSx\Ldap\Protocol\Factory\HandlerId;
 use FreeDSx\Ldap\Protocol\Factory\HandlerRouteResolverInterface;
 use FreeDSx\Ldap\Protocol\LdapMessageRequest;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Protocol\ServerProtocolHandler\ServerMonitorHandler;
 use FreeDSx\Ldap\Server\AccessControl\AccessControlInterface;
 use FreeDSx\Ldap\Server\AccessControl\Rule\AttributeAccess;
@@ -36,7 +37,6 @@
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
 use FreeDSx\Ldap\Server\Token\TokenInterface;
 
 /**
@@ -64,7 +64,7 @@ public function __construct(
     public function process(
         ServerRequestContext $context,
         MiddlewareHandlerInterface $next,
-    ): OperationResult {
+    ): ResponseStream {
         $this->authorizePrivilegedExtendedOperation($context);
 
         $routeId = $this->routeResolver->routeIdFor(
diff --git a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/ChainStep.php b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/ChainStep.php
index 845b91c5..1f178ae1 100644
--- a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/ChainStep.php
+++ b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/ChainStep.php
@@ -13,7 +13,7 @@
 
 namespace FreeDSx\Ldap\Server\Middleware\Pipeline;
 
-use FreeDSx\Ldap\Server\Operation\OperationResult;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 
 /**
  * Binds a single middleware to the next handler in the chain.
@@ -28,7 +28,7 @@ public function __construct(
         private MiddlewareHandlerInterface $next,
     ) {}
 
-    public function handle(ServerRequestContext $context): OperationResult
+    public function handle(ServerRequestContext $context): ResponseStream
     {
         return $this->middleware->process(
             $context,
diff --git a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/HandlerInvoker.php b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/HandlerInvoker.php
index 58bc4700..825254f2 100644
--- a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/HandlerInvoker.php
+++ b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/HandlerInvoker.php
@@ -14,23 +14,19 @@
 namespace FreeDSx\Ldap\Server\Middleware\Pipeline;
 
 use FreeDSx\Ldap\Protocol\Factory\ProtocolHandlerProviderInterface;
-use FreeDSx\Ldap\Protocol\Queue\Response\ResponseWriter;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 
 /**
- * Terminal handler that resolves the per-request handler, writes its response, and returns its outcome.
+ * Terminal handler that resolves and invokes the per-request protocol handler, returning its response stream.
  *
  * @internal
  * @author Chad Sikorra <Chad.Sikorra@gmail.com>
  */
 final readonly class HandlerInvoker implements MiddlewareHandlerInterface
 {
-    public function __construct(
-        private ProtocolHandlerProviderInterface $protocolHandlerProvider,
-        private ResponseWriter $responseWriter,
-    ) {}
+    public function __construct(private ProtocolHandlerProviderInterface $protocolHandlerProvider) {}
 
-    public function handle(ServerRequestContext $context): OperationResult
+    public function handle(ServerRequestContext $context): ResponseStream
     {
         $handler = $this->protocolHandlerProvider->get(
             $context->message->getRequest(),
@@ -38,14 +34,9 @@ public function handle(ServerRequestContext $context): OperationResult
             $context->searchLimits(),
         );
 
-        $stream = $handler->handleRequest(
+        return $handler->handleRequest(
             $context->message,
             $context->tokenOrFail(),
         );
-
-        return $this->responseWriter->write(
-            $stream,
-            $context->message->getMessageId(),
-        );
     }
 }
diff --git a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareChain.php b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareChain.php
index 5743543f..e0e05437 100644
--- a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareChain.php
+++ b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareChain.php
@@ -13,7 +13,7 @@
 
 namespace FreeDSx\Ldap\Server\Middleware\Pipeline;
 
-use FreeDSx\Ldap\Server\Operation\OperationResult;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 
 /**
  * Nests an ordered list of middleware around a terminal handler.
@@ -43,7 +43,7 @@ public function __construct(
         $this->pipeline = $pipeline;
     }
 
-    public function handle(ServerRequestContext $context): OperationResult
+    public function handle(ServerRequestContext $context): ResponseStream
     {
         return $this->pipeline->handle($context);
     }
diff --git a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareHandlerInterface.php b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareHandlerInterface.php
index fe3b9c70..58c7b130 100644
--- a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareHandlerInterface.php
+++ b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareHandlerInterface.php
@@ -14,7 +14,7 @@
 namespace FreeDSx\Ldap\Server\Middleware\Pipeline;
 
 use FreeDSx\Ldap\Exception\OperationException;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Socket\Exception\ConnectionException;
 
 /**
@@ -29,5 +29,5 @@ interface MiddlewareHandlerInterface
      * @throws OperationException
      * @throws ConnectionException
      */
-    public function handle(ServerRequestContext $context): OperationResult;
+    public function handle(ServerRequestContext $context): ResponseStream;
 }
diff --git a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareInterface.php b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareInterface.php
index 953fc82e..e150aea8 100644
--- a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareInterface.php
+++ b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareInterface.php
@@ -14,7 +14,7 @@
 namespace FreeDSx\Ldap\Server\Middleware\Pipeline;
 
 use FreeDSx\Ldap\Exception\OperationException;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Socket\Exception\ConnectionException;
 
 /**
@@ -34,5 +34,5 @@ interface MiddlewareInterface
     public function process(
         ServerRequestContext $context,
         MiddlewareHandlerInterface $next,
-    ): OperationResult;
+    ): ResponseStream;
 }
diff --git a/src/FreeDSx/Ldap/Server/Middleware/ReadOnlyMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/ReadOnlyMiddleware.php
index 77953b07..1be26bfe 100644
--- a/src/FreeDSx/Ldap/Server/Middleware/ReadOnlyMiddleware.php
+++ b/src/FreeDSx/Ldap/Server/Middleware/ReadOnlyMiddleware.php
@@ -28,13 +28,12 @@
 use FreeDSx\Ldap\Operation\Response\ModifyResponse;
 use FreeDSx\Ldap\Operation\ResultCode;
 use FreeDSx\Ldap\Protocol\LdapMessageResponse;
-use FreeDSx\Ldap\Protocol\Queue\ServerQueue;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\ReplicaConfig;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
 use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
 
 use function in_array;
 
@@ -59,15 +58,12 @@
         OperationType::PasswordModify,
     ];
 
-    public function __construct(
-        private ServerQueue $queue,
-        private ReplicaConfig $replicaConfig,
-    ) {}
+    public function __construct(private ReplicaConfig $replicaConfig) {}
 
     public function process(
         ServerRequestContext $context,
         MiddlewareHandlerInterface $next,
-    ): OperationResult {
+    ): ResponseStream {
         $request = $context->message->getRequest();
 
         $isWrite = in_array(
@@ -87,12 +83,13 @@ public function process(
             );
         }
 
-        $this->queue->sendMessage($this->referralResponse(
-            $request,
-            $context->message->getMessageId(),
-        ));
-
-        return OperationOutcomeResult::failed(ResultCode::REFERRAL);
+        return ResponseStream::of(
+            [$this->referralResponse(
+                $request,
+                $context->message->getMessageId(),
+            )],
+            OperationOutcomeResult::failed(ResultCode::REFERRAL),
+        );
     }
 
     private function referralResponse(
diff --git a/src/FreeDSx/Ldap/Server/Middleware/RequestValidationMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/RequestValidationMiddleware.php
index 1ea22fa3..7abe6dcb 100644
--- a/src/FreeDSx/Ldap/Server/Middleware/RequestValidationMiddleware.php
+++ b/src/FreeDSx/Ldap/Server/Middleware/RequestValidationMiddleware.php
@@ -14,10 +14,10 @@
 namespace FreeDSx\Ldap\Server\Middleware;
 
 use FreeDSx\Ldap\Exception\RequestValidationException;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
 
 use function in_array;
 use function sprintf;
@@ -41,7 +41,7 @@ final class RequestValidationMiddleware implements MiddlewareInterface
     public function process(
         ServerRequestContext $context,
         MiddlewareHandlerInterface $next,
-    ): OperationResult {
+    ): ResponseStream {
         $messageId = $context->message->getMessageId();
 
         if ($messageId === 0) {
diff --git a/src/FreeDSx/Ldap/Server/Middleware/ResourceLimitMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/ResourceLimitMiddleware.php
index 5bbfa508..1185aea4 100644
--- a/src/FreeDSx/Ldap/Server/Middleware/ResourceLimitMiddleware.php
+++ b/src/FreeDSx/Ldap/Server/Middleware/ResourceLimitMiddleware.php
@@ -13,7 +13,7 @@
 
 namespace FreeDSx\Ldap\Server\Middleware;
 
-use FreeDSx\Ldap\Server\Operation\OperationResult;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
@@ -31,7 +31,7 @@ public function __construct(
     public function process(
         ServerRequestContext $context,
         MiddlewareHandlerInterface $next,
-    ): OperationResult {
+    ): ResponseStream {
         $limits = $this->searchLimitResolver->resolve($context->tokenOrFail());
 
         return $next->handle($context->withSearchLimits($limits));
diff --git a/src/FreeDSx/Ldap/Server/Middleware/OperationErrorMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/ResponseWriterMiddleware.php
similarity index 51%
rename from src/FreeDSx/Ldap/Server/Middleware/OperationErrorMiddleware.php
rename to src/FreeDSx/Ldap/Server/Middleware/ResponseWriterMiddleware.php
index be637531..4884dc83 100644
--- a/src/FreeDSx/Ldap/Server/Middleware/OperationErrorMiddleware.php
+++ b/src/FreeDSx/Ldap/Server/Middleware/ResponseWriterMiddleware.php
@@ -13,59 +13,78 @@
 
 namespace FreeDSx\Ldap\Server\Middleware;
 
-use FreeDSx\Asn1\Exception\EncoderException;
 use FreeDSx\Ldap\Exception\OperationException;
 use FreeDSx\Ldap\Protocol\Factory\ResponseFactory;
-use FreeDSx\Ldap\Protocol\Queue\ServerQueue;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseWriter;
 use FreeDSx\Ldap\Protocol\ServerProtocolHandler\MatchedDnAccessFilterTrait;
 use FreeDSx\Ldap\Server\AccessControl\AccessControlInterface;
 use FreeDSx\Ldap\Server\Backend\LdapBackendInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
-use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
+use FreeDSx\Ldap\Server\Operation\FailedOperationResult;
 
 /**
- * Translates an OperationException thrown anywhere below into the matching response.
+ * The single sink: drains the response stream to the queue, rendering any thrown OperationException as its response.
+ *
+ * A streaming handler runs during the drain, so mid-stream failures are answered here too (a partial stream is followed by its error terminal).
  *
  * @internal
  * @author Chad Sikorra <Chad.Sikorra@gmail.com>
  */
-final readonly class OperationErrorMiddleware implements MiddlewareInterface
+final readonly class ResponseWriterMiddleware implements MiddlewareInterface
 {
     use MatchedDnAccessFilterTrait;
 
     public function __construct(
-        private ServerQueue $queue,
+        private ResponseWriter $writer,
         private LdapBackendInterface $backend,
         private AccessControlInterface $accessControl,
         private ResponseFactory $responseFactory = new ResponseFactory(),
     ) {}
 
-    /**
-     * @throws EncoderException
-     */
     public function process(
         ServerRequestContext $context,
         MiddlewareHandlerInterface $next,
-    ): OperationResult {
+    ): ResponseStream {
+        $messageId = $context->message->getMessageId();
+
         try {
-            return $next->handle($context);
+            $outcome = $this->writer->write(
+                $next->handle($context),
+                $messageId,
+            );
         } catch (OperationException $e) {
-            $this->queue->sendMessage($this->responseFactory->getStandardResponse(
+            $outcome = $this->writer->write(
+                $this->errorStream($context, $e),
+                $messageId,
+            );
+        }
+
+        return ResponseStream::resolved($outcome);
+    }
+
+    private function errorStream(
+        ServerRequestContext $context,
+        OperationException $exception,
+    ): ResponseStream {
+        return ResponseStream::of(
+            [$this->responseFactory->getStandardResponse(
                 $context->message,
-                $e->getCode(),
-                $e->getMessage(),
+                $exception->getCode(),
+                $exception->getMessage(),
                 $this->filterMatchedDn(
-                    $e->getMatchedDn(),
+                    $exception->getMatchedDn(),
                     $context->tokenOrFail(),
                     $this->backend,
                     $this->accessControl,
                 ),
-            ));
-
-            return OperationOutcomeResult::failed($e->getCode());
-        }
+            )],
+            new FailedOperationResult(
+                $context->message,
+                $exception,
+            ),
+        );
     }
 }
diff --git a/src/FreeDSx/Ldap/Server/Operation/FailedOperationResult.php b/src/FreeDSx/Ldap/Server/Operation/FailedOperationResult.php
new file mode 100644
index 00000000..dae8c107
--- /dev/null
+++ b/src/FreeDSx/Ldap/Server/Operation/FailedOperationResult.php
@@ -0,0 +1,87 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * This file is part of the FreeDSx LDAP package.
+ *
+ * (c) Chad Sikorra <Chad.Sikorra@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace FreeDSx\Ldap\Server\Operation;
+
+use FreeDSx\Ldap\Exception\OperationException;
+use FreeDSx\Ldap\Exception\SchemaRuleException;
+use FreeDSx\Ldap\Operation\Request\CompareRequest;
+use FreeDSx\Ldap\Operation\Request\SearchRequest;
+use FreeDSx\Ldap\Protocol\LdapMessageRequest;
+use FreeDSx\Ldap\Server\Logging\OperationAuditor;
+use FreeDSx\Ldap\Server\Token\TokenInterface;
+
+/**
+ * The outcome for an operation the response writer answered from a caught exception; audits by request type.
+ *
+ * @internal
+ * @author Chad Sikorra <Chad.Sikorra@gmail.com>
+ */
+final readonly class FailedOperationResult implements AuditableResult
+{
+    public function __construct(
+        private LdapMessageRequest $message,
+        private OperationException $exception,
+    ) {}
+
+    public function outcome(): OperationOutcome
+    {
+        return OperationOutcome::Failed;
+    }
+
+    public function resultCode(): int
+    {
+        return $this->exception->getCode();
+    }
+
+    public function record(
+        OperationAuditor $auditor,
+        TokenInterface $token,
+    ): void {
+        if ($this->exception instanceof SchemaRuleException) {
+            $auditor->recordSchemaViolations(
+                $this->exception->getViolations(),
+                $this->message,
+                $token,
+            );
+        }
+
+        $request = $this->message->getRequest();
+
+        if ($request instanceof SearchRequest) {
+            $auditor->recordSearchFailure(
+                $this->message,
+                $this->exception,
+                $token,
+            );
+
+            return;
+        }
+
+        if ($request instanceof CompareRequest) {
+            $auditor->recordCompareFailure(
+                $this->message,
+                $this->exception,
+                $token,
+            );
+
+            return;
+        }
+
+        $auditor->recordFailure(
+            $this->message,
+            $this->exception,
+            $token,
+        );
+    }
+}
diff --git a/src/FreeDSx/Ldap/Server/Proxy/ProxyRequestForwarder.php b/src/FreeDSx/Ldap/Server/Proxy/ProxyRequestForwarder.php
index 94701e76..3b718d23 100644
--- a/src/FreeDSx/Ldap/Server/Proxy/ProxyRequestForwarder.php
+++ b/src/FreeDSx/Ldap/Server/Proxy/ProxyRequestForwarder.php
@@ -28,11 +28,11 @@
 use FreeDSx\Ldap\Protocol\Factory\ResponseFactory;
 use FreeDSx\Ldap\Protocol\LdapMessageRequest;
 use FreeDSx\Ldap\Protocol\LdapMessageResponse;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Protocol\Queue\ServerQueue;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
 use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
 
 /**
  * Relays a request to the upstream connection and relays the response back to the client.
@@ -50,7 +50,7 @@ public function __construct(
     /**
      * {@inheritDoc}
      */
-    public function handle(ServerRequestContext $context): OperationResult
+    public function handle(ServerRequestContext $context): ResponseStream
     {
         $message = $context->message;
         $request = $message->getRequest();
@@ -59,12 +59,12 @@ public function handle(ServerRequestContext $context): OperationResult
             $this->client->unbind();
             $this->queue->close();
 
-            return OperationOutcomeResult::succeeded();
+            return ResponseStream::resolved(OperationOutcomeResult::succeeded());
         }
 
         // Synchronous, sequential forwarding means nothing is ever in flight upstream to abandon.
         if ($request instanceof AbandonRequest) {
-            return OperationOutcomeResult::succeeded();
+            return ResponseStream::resolved(OperationOutcomeResult::succeeded());
         }
 
         try {
@@ -80,7 +80,7 @@ public function handle(ServerRequestContext $context): OperationResult
                 $e->getMatchedDn(),
             ));
 
-            return OperationOutcomeResult::failed($e->getCode());
+            return ResponseStream::resolved(OperationOutcomeResult::failed($e->getCode()));
         }
 
         if ($request instanceof SearchRequest) {
@@ -89,7 +89,7 @@ public function handle(ServerRequestContext $context): OperationResult
             $this->relaySingle($message, $response);
         }
 
-        return OperationOutcomeResult::succeeded();
+        return ResponseStream::resolved(OperationOutcomeResult::succeeded());
     }
 
     /**
diff --git a/src/FreeDSx/Ldap/Server/Proxy/ProxyRequestPipeline.php b/src/FreeDSx/Ldap/Server/Proxy/ProxyRequestPipeline.php
index e3270c4f..61800224 100644
--- a/src/FreeDSx/Ldap/Server/Proxy/ProxyRequestPipeline.php
+++ b/src/FreeDSx/Ldap/Server/Proxy/ProxyRequestPipeline.php
@@ -14,11 +14,11 @@
 namespace FreeDSx\Ldap\Server\Proxy;
 
 use FreeDSx\Ldap\Operation\Request\ExtendedRequest;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Protocol\Queue\Response\ResponseWriter;
 use FreeDSx\Ldap\Protocol\ServerProtocolHandler\ServerProtocolHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
 
 /**
  * Proxy request pipeline: handles StartTLS locally and forwards everything else.
@@ -36,7 +36,7 @@ public function __construct(
     /**
      * {@inheritDoc}
      */
-    public function handle(ServerRequestContext $context): OperationResult
+    public function handle(ServerRequestContext $context): ResponseStream
     {
         $request = $context->message->getRequest();
 
@@ -46,10 +46,10 @@ public function handle(ServerRequestContext $context): OperationResult
                 $context->tokenOrFail(),
             );
 
-            return $this->responseWriter->write(
+            return ResponseStream::resolved($this->responseWriter->write(
                 $stream,
                 $context->message->getMessageId(),
-            );
+            ));
         }
 
         return $this->forwarder->handle($context);
diff --git a/src/FreeDSx/Ldap/Sync/Provider/SyncPersistStreamer.php b/src/FreeDSx/Ldap/Sync/Provider/SyncPersistStreamer.php
index 466a64ab..fb6b9f0c 100644
--- a/src/FreeDSx/Ldap/Sync/Provider/SyncPersistStreamer.php
+++ b/src/FreeDSx/Ldap/Sync/Provider/SyncPersistStreamer.php
@@ -16,15 +16,13 @@
 use FreeDSx\Ldap\Control\Sync\SyncDoneControl;
 use FreeDSx\Ldap\Entry\Dn;
 use FreeDSx\Ldap\Operation\LdapResult;
-use FreeDSx\Ldap\Operation\Request\CancelRequest;
 use FreeDSx\Ldap\Operation\Request\SearchRequest;
 use FreeDSx\Ldap\Operation\Response\ExtendedResponse;
 use FreeDSx\Ldap\Operation\Response\SearchResultDone;
 use FreeDSx\Ldap\Operation\Response\SyncInfo\SyncNewCookie;
 use FreeDSx\Ldap\Operation\ResultCode;
-use FreeDSx\Ldap\Protocol\LdapMessageRequest;
 use FreeDSx\Ldap\Protocol\LdapMessageResponse;
-use FreeDSx\Ldap\Protocol\Queue\ServerQueue;
+use FreeDSx\Ldap\Protocol\Queue\Response\Cancellation;
 use FreeDSx\Ldap\Server\Backend\LdapBackendInterface;
 use FreeDSx\Ldap\Server\Backend\Storage\Journal\Change\ChangeType;
 use FreeDSx\Ldap\Server\Backend\Storage\Journal\Change\PendingChange;
@@ -33,9 +31,10 @@
 use FreeDSx\Ldap\Server\Backend\Storage\Journal\Read\ChangeStream;
 use FreeDSx\Ldap\Server\Clock\Sleeper\SleeperInterface;
 use FreeDSx\Ldap\Server\Token\TokenInterface;
+use Generator;
 
 /**
- * Tails the change journal for a refreshAndPersist consumer, pushing changes until it cancels or disconnects.
+ * Tails the change journal for a refreshAndPersist consumer, yielding changes until it cancels or disconnects.
  *
  * @author Chad Sikorra <Chad.Sikorra@gmail.com>
  */
@@ -47,7 +46,6 @@
     public const DEFAULT_POLL_INTERVAL = 1.0;
 
     public function __construct(
-        private ServerQueue $queue,
         private LdapBackendInterface $backend,
         private SyncResultProjector $projector,
         private ChangeStream $stream,
@@ -84,15 +82,20 @@ public function projectSince(
     }
 
     /**
-     * Runs the persist phase: pushes each change as it lands, advancing the cookie, until cancel or disconnect.
+     * The persist phase: yields each change as it lands, advancing the cookie, until cancel or disconnect.
+     *
+     * The writer polls the queue and offers any abandon/cancel into the shared token, which this loop reads.
+     *
+     * @return Generator<LdapMessageResponse>
      */
-    public function persist(
+    public function stream(
         int $startSeq,
         SearchRequest $request,
         TokenInterface $token,
         int $messageId,
         Dn $baseDn,
-    ): void {
+        Cancellation $cancellation,
+    ): Generator {
         $lastSeq = $startSeq;
         $origin = $this->stream->origin();
 
@@ -101,26 +104,35 @@ public function persist(
 
             // A trim past the consumer's position would silently drop changes: force a full re-sync instead.
             if ($latestSeq > $lastSeq && !$this->stream->retainsSince($lastSeq)) {
-                $this->sendRefreshRequired($messageId, $baseDn, $origin, $latestSeq);
+                yield $this->refreshRequired($messageId, $baseDn, $origin, $latestSeq);
 
                 return;
             }
 
             if ($latestSeq > $lastSeq) {
-                $this->pushChanges($lastSeq, $request, $token, $messageId);
+                yield from $this->pushChanges($lastSeq, $request, $token, $messageId);
                 $lastSeq = $latestSeq;
             }
 
             // Advances the client cookie, and doubles as a keepalive that surfaces a dead peer on the next poll.
-            $this->queue->sendMessage(new LdapMessageResponse(
+            yield new LdapMessageResponse(
                 $messageId,
                 new SyncNewCookie((new SyncCookie($origin, $lastSeq))->encode()),
-            ));
+            );
 
-            $signal = $this->queue->peekForCancelSignal($messageId);
+            $signal = $cancellation->signal();
 
             if ($signal !== null) {
-                $this->sendTerminal($signal, $messageId, $baseDn, $origin, $lastSeq);
+                // Abandon carries no response (RFC 4511 §4.11); only a Cancel gets an acknowledged close.
+                if ($cancellation->isCanceled()) {
+                    yield from $this->terminal(
+                        $signal->getMessageId(),
+                        $messageId,
+                        $baseDn,
+                        $origin,
+                        $lastSeq,
+                    );
+                }
 
                 return;
             }
@@ -129,18 +141,21 @@ public function persist(
         }
     }
 
+    /**
+     * @return Generator<LdapMessageResponse>
+     */
     private function pushChanges(
         int $sinceSeq,
         SearchRequest $request,
         TokenInterface $token,
         int $messageId,
-    ): void {
+    ): Generator {
         foreach ($this->projectSince($sinceSeq, $request, $token) as $result) {
-            $this->queue->sendMessage(new LdapMessageResponse(
+            yield new LdapMessageResponse(
                 $messageId,
                 $result->entry,
                 $result->control,
-            ));
+            );
         }
     }
 
@@ -173,44 +188,40 @@ private function scopeFor(SearchRequest $request): ChangeScope
         };
     }
 
-    private function sendTerminal(
-        LdapMessageRequest $signal,
+    /**
+     * @return Generator<LdapMessageResponse>
+     */
+    private function terminal(
+        int $cancelMessageId,
         int $messageId,
         Dn $baseDn,
         ReplicaId $origin,
         int $lastSeq,
-    ): void {
-        // Abandon carries no response (RFC 4511 §4.11); only a Cancel gets an acknowledged close.
-        if (!($signal->getRequest() instanceof CancelRequest)) {
-            return;
-        }
-
-        $this->queue->sendMessage(
-            new LdapMessageResponse(
-                $messageId,
-                new SearchResultDone(
-                    ResultCode::CANCELED,
-                    $baseDn->toString(),
-                ),
-                new SyncDoneControl(
-                    (new SyncCookie($origin, $lastSeq))->encode(),
-                    true,
-                ),
+    ): Generator {
+        yield new LdapMessageResponse(
+            $messageId,
+            new SearchResultDone(
+                ResultCode::CANCELED,
+                $baseDn->toString(),
             ),
-            new LdapMessageResponse(
-                $signal->getMessageId(),
-                new ExtendedResponse(new LdapResult(ResultCode::SUCCESS)),
+            new SyncDoneControl(
+                (new SyncCookie($origin, $lastSeq))->encode(),
+                true,
             ),
         );
+        yield new LdapMessageResponse(
+            $cancelMessageId,
+            new ExtendedResponse(new LdapResult(ResultCode::SUCCESS)),
+        );
     }
 
-    private function sendRefreshRequired(
+    private function refreshRequired(
         int $messageId,
         Dn $baseDn,
         ReplicaId $origin,
         int $latestSeq,
-    ): void {
-        $this->queue->sendMessage(new LdapMessageResponse(
+    ): LdapMessageResponse {
+        return new LdapMessageResponse(
             $messageId,
             new SearchResultDone(
                 ResultCode::SYNCHRONIZATION_REFRESH_REQUIRED,
@@ -220,6 +231,6 @@ private function sendRefreshRequired(
                 (new SyncCookie($origin, $latestSeq))->encode(),
                 true,
             ),
-        ));
+        );
     }
 }
diff --git a/tests/support/Middleware/CallbackMiddlewareHandler.php b/tests/support/Middleware/CallbackMiddlewareHandler.php
index 16b2a1e0..ba583c9e 100644
--- a/tests/support/Middleware/CallbackMiddlewareHandler.php
+++ b/tests/support/Middleware/CallbackMiddlewareHandler.php
@@ -14,6 +14,7 @@
 namespace Tests\Support\FreeDSx\Ldap\Middleware;
 
 use Closure;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
 use FreeDSx\Ldap\Server\Operation\OperationResult;
@@ -30,8 +31,8 @@
      */
     public function __construct(private Closure $callback) {}
 
-    public function handle(ServerRequestContext $context): OperationResult
+    public function handle(ServerRequestContext $context): ResponseStream
     {
-        return ($this->callback)($context);
+        return ResponseStream::resolved(($this->callback)($context));
     }
 }
diff --git a/tests/support/Middleware/RecordingMiddleware.php b/tests/support/Middleware/RecordingMiddleware.php
index d80920d0..e3f6cfae 100644
--- a/tests/support/Middleware/RecordingMiddleware.php
+++ b/tests/support/Middleware/RecordingMiddleware.php
@@ -13,10 +13,10 @@
 
 namespace Tests\Support\FreeDSx\Ldap\Middleware;
 
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
 
 /**
  * Records before/after markers around the next handler for ordering assertions.
@@ -33,7 +33,7 @@ public function __construct(
     public function process(
         ServerRequestContext $context,
         MiddlewareHandlerInterface $next,
-    ): OperationResult {
+    ): ResponseStream {
         $this->log->record('before:' . $this->label);
         $result = $next->handle($context);
         $this->log->record('after:' . $this->label);
diff --git a/tests/support/Middleware/RecordingMiddlewareHandler.php b/tests/support/Middleware/RecordingMiddlewareHandler.php
index 4d9688f0..e87152cf 100644
--- a/tests/support/Middleware/RecordingMiddlewareHandler.php
+++ b/tests/support/Middleware/RecordingMiddlewareHandler.php
@@ -13,10 +13,10 @@
 
 namespace Tests\Support\FreeDSx\Ldap\Middleware;
 
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
 use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
 
 /**
  * Terminal handler that records its invocation and captures the received context.
@@ -32,11 +32,11 @@ public function __construct(
         private readonly string $label = 'terminal',
     ) {}
 
-    public function handle(ServerRequestContext $context): OperationResult
+    public function handle(ServerRequestContext $context): ResponseStream
     {
         $this->received = $context;
         $this->log->record($this->label);
 
-        return OperationOutcomeResult::succeeded();
+        return ResponseStream::resolved(OperationOutcomeResult::succeeded());
     }
 }
diff --git a/tests/support/Middleware/StreamMiddlewareHandler.php b/tests/support/Middleware/StreamMiddlewareHandler.php
new file mode 100644
index 00000000..e4c68266
--- /dev/null
+++ b/tests/support/Middleware/StreamMiddlewareHandler.php
@@ -0,0 +1,33 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * This file is part of the FreeDSx LDAP package.
+ *
+ * (c) Chad Sikorra <Chad.Sikorra@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Tests\Support\FreeDSx\Ldap\Middleware;
+
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
+use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
+use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
+
+/**
+ * Terminal handler that returns a preconfigured response stream (e.g. a streaming one that fails mid-drain).
+ *
+ * @author Chad Sikorra <Chad.Sikorra@gmail.com>
+ */
+final readonly class StreamMiddlewareHandler implements MiddlewareHandlerInterface
+{
+    public function __construct(private ResponseStream $stream) {}
+
+    public function handle(ServerRequestContext $context): ResponseStream
+    {
+        return $this->stream;
+    }
+}
diff --git a/tests/support/Middleware/StubMiddlewareHandler.php b/tests/support/Middleware/StubMiddlewareHandler.php
index 991a61b5..b9484991 100644
--- a/tests/support/Middleware/StubMiddlewareHandler.php
+++ b/tests/support/Middleware/StubMiddlewareHandler.php
@@ -13,12 +13,13 @@
 
 namespace Tests\Support\FreeDSx\Ldap\Middleware;
 
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
 use FreeDSx\Ldap\Server\Operation\OperationResult;
 
 /**
- * Terminal handler that returns a preconfigured result.
+ * Terminal handler that resolves to a preconfigured outcome.
  *
  * @author Chad Sikorra <Chad.Sikorra@gmail.com>
  */
@@ -26,8 +27,8 @@
 {
     public function __construct(private OperationResult $result) {}
 
-    public function handle(ServerRequestContext $context): OperationResult
+    public function handle(ServerRequestContext $context): ResponseStream
     {
-        return $this->result;
+        return ResponseStream::resolved($this->result);
     }
 }
diff --git a/tests/support/Middleware/ThrowingMiddleware.php b/tests/support/Middleware/ThrowingMiddleware.php
index 29a5ff21..63fe33ed 100644
--- a/tests/support/Middleware/ThrowingMiddleware.php
+++ b/tests/support/Middleware/ThrowingMiddleware.php
@@ -13,10 +13,10 @@
 
 namespace Tests\Support\FreeDSx\Ldap\Middleware;
 
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
 use Throwable;
 
 /**
@@ -31,7 +31,7 @@ public function __construct(private Throwable $throwable) {}
     public function process(
         ServerRequestContext $context,
         MiddlewareHandlerInterface $next,
-    ): OperationResult {
+    ): ResponseStream {
         throw $this->throwable;
     }
 }
diff --git a/tests/support/Middleware/ThrowingMiddlewareHandler.php b/tests/support/Middleware/ThrowingMiddlewareHandler.php
index 20245bcf..4669c311 100644
--- a/tests/support/Middleware/ThrowingMiddlewareHandler.php
+++ b/tests/support/Middleware/ThrowingMiddlewareHandler.php
@@ -13,9 +13,9 @@
 
 namespace Tests\Support\FreeDSx\Ldap\Middleware;
 
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
-use FreeDSx\Ldap\Server\Operation\OperationResult;
 use Throwable;
 
 /**
@@ -27,7 +27,7 @@
 {
     public function __construct(private Throwable $throwable) {}
 
-    public function handle(ServerRequestContext $context): OperationResult
+    public function handle(ServerRequestContext $context): ResponseStream
     {
         throw $this->throwable;
     }
diff --git a/tests/unit/Protocol/Factory/ProtocolHandlerProviderTest.php b/tests/unit/Protocol/Factory/ProtocolHandlerProviderTest.php
index 7865d5fc..1bed9ab6 100644
--- a/tests/unit/Protocol/Factory/ProtocolHandlerProviderTest.php
+++ b/tests/unit/Protocol/Factory/ProtocolHandlerProviderTest.php
@@ -25,7 +25,7 @@
 use FreeDSx\Ldap\Protocol\Factory\ProtocolHandlerFactoryMap;
 use FreeDSx\Ldap\Protocol\Factory\ProtocolHandlerProvider;
 use FreeDSx\Ldap\Protocol\Factory\ServerProtocolHandlerFactory;
-use FreeDSx\Ldap\Protocol\Queue\ServerQueue;
+use FreeDSx\Ldap\Protocol\Queue\ConnectionControl;
 use FreeDSx\Ldap\Protocol\ServerProtocolHandler\ServerDispatchHandler;
 use FreeDSx\Ldap\Protocol\ServerProtocolHandler\ServerPagingHandler;
 use FreeDSx\Ldap\Protocol\ServerProtocolHandler\ServerPasswordModifyHandler;
@@ -252,7 +252,7 @@ public function test_it_throws_when_password_policy_is_enabled_but_the_backend_i
     private function handlerContext(?PasswordPolicyContext $passwordPolicyContext = null): HandlerContext
     {
         return new HandlerContext(
-            queue: $this->createMock(ServerQueue::class),
+            connection: $this->createMock(ConnectionControl::class),
             eventLogger: new EventLogger(null),
             requestHistory: new RequestHistory(),
             passwordPolicyContext: $passwordPolicyContext,
diff --git a/tests/unit/Protocol/ServerProtocolHandler/ServerSyncHandlerTest.php b/tests/unit/Protocol/ServerProtocolHandler/ServerSyncHandlerTest.php
index 70d62a99..a032d94b 100644
--- a/tests/unit/Protocol/ServerProtocolHandler/ServerSyncHandlerTest.php
+++ b/tests/unit/Protocol/ServerProtocolHandler/ServerSyncHandlerTest.php
@@ -29,6 +29,7 @@
 use FreeDSx\Ldap\Protocol\Authorization\AuthzId;
 use FreeDSx\Ldap\Protocol\LdapMessageRequest;
 use FreeDSx\Ldap\Protocol\LdapMessageResponse;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseWriter;
 use FreeDSx\Ldap\Protocol\Queue\ServerQueue;
 use FreeDSx\Ldap\Protocol\ServerProtocolHandler\ServerSyncHandler;
 use FreeDSx\Ldap\Search\Filters;
@@ -148,15 +149,6 @@ protected function setUp(): void
                 return $this->queue;
             });
         $this->queue
-            ->method('sendMessage')
-            ->willReturnCallback(function (LdapMessageResponse ...$responses): ServerQueue {
-                foreach ($responses as $response) {
-                    $this->sent[] = $response;
-                }
-
-                return $this->queue;
-            });
-        $this->queue
             ->method('peekForCancelSignal')
             ->willReturnCallback(fn(): ?LdapMessageRequest => array_shift($this->cancelSignals));
 
@@ -166,7 +158,6 @@ protected function setUp(): void
             filterEvaluator: $this->filterEvaluator,
         );
         $this->persistStreamer = new SyncPersistStreamer(
-            queue: $this->queue,
             backend: $this->backend,
             projector: $this->syncProjector,
             stream: $this->changeStream,
@@ -174,7 +165,6 @@ protected function setUp(): void
         );
 
         $this->subject = new ServerSyncHandler(
-            queue: $this->queue,
             backend: $this->backend,
             projector: $this->syncProjector,
             changeStream: $this->changeStream,
@@ -459,9 +449,9 @@ public function test_refresh_and_persist_full_refresh_ends_the_refresh_phase_wit
         // A pending cancel terminates the persist loop after the boundary is emitted.
         $this->cancelSignals = [new LdapMessageRequest(9, new CancelRequest(1))];
 
-        $this->persistHandler()->handleRequest(
+        $this->drive(
+            $this->persistHandler(),
             $this->syncMessage(null, SyncRequestControl::MODE_REFRESH_AND_PERSIST),
-            $this->token,
         );
 
         $boundary = null;
@@ -483,7 +473,6 @@ public function test_refresh_and_persist_full_refresh_ends_the_refresh_phase_wit
     public function test_sync_is_declined_when_no_change_stream_is_available(): void
     {
         $handler = new ServerSyncHandler(
-            queue: $this->queue,
             backend: $this->backend,
             projector: new SyncResultProjector(
                 accessControl: $this->accessControl,
@@ -503,7 +492,6 @@ public function test_sync_is_declined_when_no_change_stream_is_available(): void
     private function persistHandler(): ServerSyncHandler
     {
         return new ServerSyncHandler(
-            queue: $this->queue,
             backend: $this->backend,
             projector: $this->syncProjector,
             changeStream: $this->changeStream,
@@ -517,10 +505,26 @@ private function handle(
         int $mode = SyncRequestControl::MODE_REFRESH_ONLY,
         bool $reloadHint = false,
     ): OperationResult {
-        return $this->subject->handleRequest(
+        return $this->drive(
+            $this->subject,
             $this->syncMessage($cookie, $mode, $reloadHint),
-            $this->token,
-        )->outcome();
+        );
+    }
+
+    /**
+     * Drives the handler's stream through the writer, which populates the queue and polls for cancellation.
+     */
+    private function drive(
+        ServerSyncHandler $handler,
+        LdapMessageRequest $message,
+    ): OperationResult {
+        return (new ResponseWriter($this->queue))->write(
+            $handler->handleRequest(
+                $message,
+                $this->token,
+            ),
+            $message->getMessageId(),
+        );
     }
 
     private function syncMessage(
diff --git a/tests/unit/Server/Middleware/BindMiddlewareTest.php b/tests/unit/Server/Middleware/BindMiddlewareTest.php
index 8fff69cb..a68f16eb 100644
--- a/tests/unit/Server/Middleware/BindMiddlewareTest.php
+++ b/tests/unit/Server/Middleware/BindMiddlewareTest.php
@@ -89,7 +89,7 @@ public function test_a_successful_bind_stores_the_token_and_short_circuits(): vo
         );
         self::assertSame(
             OperationOutcome::Succeeded,
-            $result->outcome(),
+            $result->outcome()->outcome(),
         );
     }
 
diff --git a/tests/unit/Server/Middleware/OperationAuditMiddlewareTest.php b/tests/unit/Server/Middleware/OperationAuditMiddlewareTest.php
index cce64f4e..41d32c81 100644
--- a/tests/unit/Server/Middleware/OperationAuditMiddlewareTest.php
+++ b/tests/unit/Server/Middleware/OperationAuditMiddlewareTest.php
@@ -29,6 +29,7 @@
 use FreeDSx\Ldap\Server\Logging\OperationAuditor;
 use FreeDSx\Ldap\Server\Middleware\OperationAuditMiddleware;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
+use FreeDSx\Ldap\Server\Operation\FailedOperationResult;
 use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult;
 use FreeDSx\Ldap\Server\Operation\WriteOperationResult;
 use FreeDSx\Ldap\Server\Token\BindToken;
@@ -89,7 +90,7 @@ public function test_it_does_not_record_a_non_auditable_result(): void
         );
     }
 
-    public function test_it_returns_the_result_from_the_next_handler_unchanged(): void
+    public function test_it_returns_the_stream_from_the_next_handler_unchanged(): void
     {
         $result = OperationOutcomeResult::failed();
 
@@ -98,44 +99,48 @@ public function test_it_returns_the_result_from_the_next_handler_unchanged(): vo
             $this->subject->process(
                 $this->context,
                 new StubMiddlewareHandler($result),
-            ),
+            )->outcome(),
         );
     }
 
-    public function test_it_audits_a_write_authorization_denial_and_rethrows(): void
+    public function test_it_audits_a_write_authorization_denial(): void
     {
-        $exception = new OperationException(
-            'Denied.',
-            ResultCode::INSUFFICIENT_ACCESS_RIGHTS,
+        $this->auditFailure(
+            $this->context,
+            new OperationException(
+                'Denied.',
+                ResultCode::INSUFFICIENT_ACCESS_RIGHTS,
+            ),
         );
 
-        $this->processThrowing($this->context, $exception);
-
         self::assertTrue($this->wasLogged('authz.denied.write'));
     }
 
     public function test_it_audits_a_search_authorization_denial(): void
     {
         $context = $this->contextFor((new SearchRequest(Filters::present('cn')))->base('dc=bar'));
-        $exception = new OperationException(
-            'Denied.',
-            ResultCode::INSUFFICIENT_ACCESS_RIGHTS,
-        );
 
-        $this->processThrowing($context, $exception);
+        $this->auditFailure(
+            $context,
+            new OperationException(
+                'Denied.',
+                ResultCode::INSUFFICIENT_ACCESS_RIGHTS,
+            ),
+        );
 
         self::assertTrue($this->wasLogged('authz.denied.read'));
     }
 
     public function test_it_audits_a_critical_control_rejection(): void
     {
-        $exception = new OperationException(
-            'Critical control 1.2.3.4 is not supported.',
-            ResultCode::UNAVAILABLE_CRITICAL_EXTENSION,
+        $this->auditFailure(
+            $this->context,
+            new OperationException(
+                'Critical control 1.2.3.4 is not supported.',
+                ResultCode::UNAVAILABLE_CRITICAL_EXTENSION,
+            ),
         );
 
-        $this->processThrowing($this->context, $exception);
-
         self::assertTrue($this->wasLogged('control.critical.rejected'));
     }
 
@@ -157,12 +162,12 @@ public function test_it_records_schema_violations_from_a_schema_rule_exception()
             $violations,
         );
 
-        $this->processThrowing($this->context, $exception);
+        $this->auditFailure($this->context, $exception);
 
         self::assertTrue($this->wasLogged('schema.violation'));
     }
 
-    public function test_it_rethrows_the_caught_operation_exception(): void
+    public function test_it_lets_an_exception_from_the_next_handler_propagate(): void
     {
         $exception = new OperationException('Boom.');
 
@@ -174,18 +179,20 @@ public function test_it_rethrows_the_caught_operation_exception(): void
         );
     }
 
-    private function processThrowing(
+    /**
+     * A failure reaches the audit middleware as a resolved FailedOperationResult, not a thrown exception.
+     */
+    private function auditFailure(
         ServerRequestContext $context,
         OperationException $exception,
     ): void {
-        try {
-            $this->subject->process(
-                $context,
-                new ThrowingMiddlewareHandler($exception),
-            );
-            self::fail('Expected the OperationException to be re-thrown.');
-        } catch (OperationException) {
-        }
+        $this->subject->process(
+            $context,
+            new StubMiddlewareHandler(new FailedOperationResult(
+                $context->message,
+                $exception,
+            )),
+        );
     }
 
     private function wasLogged(string $event): bool
diff --git a/tests/unit/Server/Middleware/ReadOnlyMiddlewareTest.php b/tests/unit/Server/Middleware/ReadOnlyMiddlewareTest.php
index 85d9c300..3919dfa2 100644
--- a/tests/unit/Server/Middleware/ReadOnlyMiddlewareTest.php
+++ b/tests/unit/Server/Middleware/ReadOnlyMiddlewareTest.php
@@ -27,12 +27,13 @@
 use FreeDSx\Ldap\Operations;
 use FreeDSx\Ldap\Protocol\LdapMessageRequest;
 use FreeDSx\Ldap\Protocol\LdapMessageResponse;
-use FreeDSx\Ldap\Protocol\Queue\ServerQueue;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
 use FreeDSx\Ldap\ReplicaConfig;
 use FreeDSx\Ldap\Search\Filters;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
 use FreeDSx\Ldap\Server\Middleware\ReadOnlyMiddleware;
+use FreeDSx\Ldap\Server\Operation\OperationOutcome;
 use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult;
 use PHPUnit\Framework\Attributes\DataProvider;
 use PHPUnit\Framework\MockObject\MockObject;
@@ -40,8 +41,6 @@
 
 final class ReadOnlyMiddlewareTest extends TestCase
 {
-    private ServerQueue&MockObject $queue;
-
     private MiddlewareHandlerInterface&MockObject $next;
 
     private ReplicaConfig $replicaConfig;
@@ -50,17 +49,13 @@ final class ReadOnlyMiddlewareTest extends TestCase
 
     protected function setUp(): void
     {
-        $this->queue = $this->createMock(ServerQueue::class);
         $this->next = $this->createMock(MiddlewareHandlerInterface::class);
         $this->replicaConfig = new ReplicaConfig(
             (new ClientOptions())
                 ->setServers(['primary.example.com'])
                 ->setPort(389),
         );
-        $this->subject = new ReadOnlyMiddleware(
-            $this->queue,
-            $this->replicaConfig,
-        );
+        $this->subject = new ReadOnlyMiddleware($this->replicaConfig);
     }
 
     /**
@@ -104,31 +99,31 @@ public function test_a_write_is_referred_to_the_primary_by_default(
             ->expects(self::never())
             ->method('handle');
 
-        $sent = null;
-        $this->queue
-            ->expects(self::once())
-            ->method('sendMessage')
-            ->willReturnCallback(function (LdapMessageResponse $message) use (&$sent): ServerQueue {
-                $sent = $message;
-
-                return $this->queue;
-            });
-
-        $result = $this->subject->process(
+        $stream = $this->subject->process(
             $this->context($request),
             $this->next,
         );
 
         self::assertSame(
             ResultCode::REFERRAL,
-            $result->resultCode(),
+            $stream->outcome()->resultCode(),
+        );
+        self::assertSame(
+            OperationOutcome::Failed,
+            $stream->outcome()->outcome(),
+        );
+
+        $messages = [...$stream->messages];
+        self::assertCount(
+            1,
+            $messages,
         );
         self::assertInstanceOf(
             LdapMessageResponse::class,
-            $sent,
+            $messages[0],
         );
 
-        $response = $sent->getResponse();
+        $response = $messages[0]->getResponse();
         self::assertInstanceOf(
             $expectedResponse,
             $response,
@@ -149,9 +144,6 @@ public function test_a_write_is_rejected_when_referral_is_disabled(): void
         $this->next
             ->expects(self::never())
             ->method('handle');
-        $this->queue
-            ->expects(self::never())
-            ->method('sendMessage');
 
         $this->expectException(OperationException::class);
         $this->expectExceptionCode(ResultCode::UNWILLING_TO_PERFORM);
@@ -165,16 +157,13 @@ public function test_a_write_is_rejected_when_referral_is_disabled(): void
     public function test_a_read_passes_through_to_the_next_handler(): void
     {
         $context = $this->context(Operations::search(Filters::present('objectClass')));
-        $expected = OperationOutcomeResult::succeeded();
+        $expected = ResponseStream::none(OperationOutcomeResult::succeeded());
 
         $this->next
             ->expects(self::once())
             ->method('handle')
             ->with($context)
             ->willReturn($expected);
-        $this->queue
-            ->expects(self::never())
-            ->method('sendMessage');
 
         self::assertSame(
             $expected,
diff --git a/tests/unit/Server/Middleware/OperationErrorMiddlewareTest.php b/tests/unit/Server/Middleware/ResponseWriterMiddlewareTest.php
similarity index 62%
rename from tests/unit/Server/Middleware/OperationErrorMiddlewareTest.php
rename to tests/unit/Server/Middleware/ResponseWriterMiddlewareTest.php
index 829f9f53..ea1d23ce 100644
--- a/tests/unit/Server/Middleware/OperationErrorMiddlewareTest.php
+++ b/tests/unit/Server/Middleware/ResponseWriterMiddlewareTest.php
@@ -18,24 +18,33 @@
 use FreeDSx\Ldap\Exception\OperationException;
 use FreeDSx\Ldap\Operation\Request\DeleteRequest;
 use FreeDSx\Ldap\Operation\Request\RequestInterface;
+use FreeDSx\Ldap\Operation\Request\SearchRequest;
 use FreeDSx\Ldap\Operation\Response\DeleteResponse;
+use FreeDSx\Ldap\Operation\Response\SearchResultDone;
+use FreeDSx\Ldap\Operation\Response\SearchResultEntry;
 use FreeDSx\Ldap\Operation\ResultCode;
 use FreeDSx\Ldap\Protocol\LdapMessageRequest;
 use FreeDSx\Ldap\Protocol\LdapMessageResponse;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseWriter;
 use FreeDSx\Ldap\Protocol\Queue\ServerQueue;
+use FreeDSx\Ldap\Search\Filters;
 use FreeDSx\Ldap\Server\AccessControl\AccessControlInterface;
 use FreeDSx\Ldap\Server\Backend\LdapBackendInterface;
-use FreeDSx\Ldap\Server\Middleware\OperationErrorMiddleware;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
+use FreeDSx\Ldap\Server\Middleware\ResponseWriterMiddleware;
 use FreeDSx\Ldap\Server\Operation\OperationOutcome;
 use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult;
+use FreeDSx\Ldap\Server\Operation\SearchOperationResult;
 use FreeDSx\Ldap\Server\Token\TokenInterface;
+use Generator;
 use PHPUnit\Framework\MockObject\MockObject;
 use PHPUnit\Framework\TestCase;
+use Tests\Support\FreeDSx\Ldap\Middleware\StreamMiddlewareHandler;
 use Tests\Support\FreeDSx\Ldap\Middleware\StubMiddlewareHandler;
 use Tests\Support\FreeDSx\Ldap\Middleware\ThrowingMiddlewareHandler;
 
-final class OperationErrorMiddlewareTest extends TestCase
+final class ResponseWriterMiddlewareTest extends TestCase
 {
     private ServerQueue&MockObject $mockQueue;
 
@@ -45,9 +54,12 @@ final class OperationErrorMiddlewareTest extends TestCase
 
     private TokenInterface&MockObject $mockToken;
 
-    private OperationErrorMiddleware $subject;
+    private ResponseWriterMiddleware $subject;
 
-    private ?LdapMessageResponse $sent = null;
+    /**
+     * @var list<LdapMessageResponse>
+     */
+    private array $sent = [];
 
     protected function setUp(): void
     {
@@ -57,40 +69,46 @@ protected function setUp(): void
         $this->mockToken = $this->createMock(TokenInterface::class);
 
         $this->mockQueue
-            ->method('sendMessage')
-            ->willReturnCallback(function (LdapMessageResponse $response): ServerQueue {
-                $this->sent = $response;
+            ->method('sendMessages')
+            ->willReturnCallback(function (iterable $responses): ServerQueue {
+                foreach ($responses as $response) {
+                    if ($response instanceof LdapMessageResponse) {
+                        $this->sent[] = $response;
+                    }
+                }
 
                 return $this->mockQueue;
             });
 
-        $this->subject = new OperationErrorMiddleware(
-            $this->mockQueue,
+        $this->subject = new ResponseWriterMiddleware(
+            new ResponseWriter($this->mockQueue),
             $this->mockBackend,
             $this->mockAccessControl,
         );
     }
 
-    public function test_it_passes_a_successful_result_through_without_sending(): void
+    public function test_it_resolves_a_successful_outcome_without_rendering_an_error(): void
     {
-        $this->mockQueue
-            ->expects(self::never())
-            ->method('sendMessage');
-
         $result = OperationOutcomeResult::succeeded();
 
+        $stream = $this->subject->process(
+            $this->contextFor(new DeleteRequest('cn=foo,dc=bar')),
+            new StubMiddlewareHandler($result),
+        );
+
         self::assertSame(
             $result,
-            $this->subject->process(
-                $this->contextFor(new DeleteRequest('cn=foo,dc=bar')),
-                new StubMiddlewareHandler($result),
-            ),
+            $stream->outcome(),
+        );
+        self::assertSame(
+            [],
+            $this->sent,
         );
     }
 
     public function test_it_renders_the_response_for_a_thrown_exception_and_does_not_rethrow(): void
     {
-        $result = $this->subject->process(
+        $stream = $this->subject->process(
             $this->contextFor(new DeleteRequest('cn=foo,dc=bar')),
             new ThrowingMiddlewareHandler(new OperationException(
                 'Unwilling.',
@@ -100,10 +118,10 @@ public function test_it_renders_the_response_for_a_thrown_exception_and_does_not
 
         self::assertSame(
             OperationOutcome::Failed,
-            $result->outcome(),
+            $stream->outcome()->outcome(),
         );
 
-        $response = $this->sent?->getResponse();
+        $response = $this->firstSent()?->getResponse();
         self::assertInstanceOf(DeleteResponse::class, $response);
         self::assertSame(
             ResultCode::UNWILLING_TO_PERFORM,
@@ -111,6 +129,35 @@ public function test_it_renders_the_response_for_a_thrown_exception_and_does_not
         );
     }
 
+    public function test_a_mid_stream_failure_is_answered_after_the_entries_already_written(): void
+    {
+        $entry = new LdapMessageResponse(1, new SearchResultEntry(Entry::create('cn=a,dc=bar')));
+
+        $this->subject->process(
+            $this->contextFor((new SearchRequest(Filters::present('cn')))->base('dc=bar')),
+            new StreamMiddlewareHandler(ResponseStream::streaming(
+                (function () use ($entry): Generator {
+                    yield $entry;
+
+                    throw new OperationException('Backend failed.', ResultCode::OPERATIONS_ERROR);
+                })(),
+                static fn(): SearchOperationResult => throw new OperationException('unreached'),
+            )),
+        );
+
+        // The already-produced entry went out, then the failure rendered as the search's terminal.
+        self::assertSame(
+            $entry,
+            $this->sent[0] ?? null,
+        );
+        $done = $this->sent[1]->getResponse() ?? null;
+        self::assertInstanceOf(SearchResultDone::class, $done);
+        self::assertSame(
+            ResultCode::OPERATIONS_ERROR,
+            $done->getResultCode(),
+        );
+    }
+
     public function test_it_keeps_a_matched_dn_the_token_may_see(): void
     {
         $this->mockBackend
@@ -125,7 +172,7 @@ public function test_it_keeps_a_matched_dn_the_token_may_see(): void
             new ThrowingMiddlewareHandler($this->noSuchObject(new Dn('dc=bar'))),
         );
 
-        $response = $this->sent?->getResponse();
+        $response = $this->firstSent()?->getResponse();
         self::assertInstanceOf(DeleteResponse::class, $response);
         self::assertSame(
             'dc=bar',
@@ -147,7 +194,7 @@ public function test_it_drops_a_matched_dn_the_access_control_hides(): void
             new ThrowingMiddlewareHandler($this->noSuchObject(new Dn('dc=bar'))),
         );
 
-        $response = $this->sent?->getResponse();
+        $response = $this->firstSent()?->getResponse();
         self::assertInstanceOf(DeleteResponse::class, $response);
         self::assertSame(
             '',
@@ -166,7 +213,7 @@ public function test_it_drops_a_matched_dn_when_the_backend_has_no_ancestor(): v
             new ThrowingMiddlewareHandler($this->noSuchObject(new Dn('dc=bar'))),
         );
 
-        $response = $this->sent?->getResponse();
+        $response = $this->firstSent()?->getResponse();
         self::assertInstanceOf(DeleteResponse::class, $response);
         self::assertSame(
             '',
@@ -184,6 +231,11 @@ private function noSuchObject(Dn $matchedDn): OperationException
         );
     }
 
+    private function firstSent(): ?LdapMessageResponse
+    {
+        return $this->sent[0] ?? null;
+    }
+
     private function contextFor(RequestInterface $request): ServerRequestContext
     {
         return new ServerRequestContext(
diff --git a/tests/unit/Server/Proxy/ProxyRequestForwarderTest.php b/tests/unit/Server/Proxy/ProxyRequestForwarderTest.php
index 5b15ca63..846be038 100644
--- a/tests/unit/Server/Proxy/ProxyRequestForwarderTest.php
+++ b/tests/unit/Server/Proxy/ProxyRequestForwarderTest.php
@@ -65,7 +65,7 @@ public function test_it_relays_a_single_response_under_the_original_message_id()
             new DeleteRequest('cn=foo,dc=bar'),
         ));
 
-        self::assertSame(OperationOutcome::Succeeded, $result->outcome());
+        self::assertSame(OperationOutcome::Succeeded, $result->outcome()->outcome());
     }
 
     public function test_it_relays_an_upstream_error_as_a_response(): void
@@ -92,7 +92,7 @@ public function test_it_relays_an_upstream_error_as_a_response(): void
             new DeleteRequest('cn=missing,dc=bar'),
         ));
 
-        self::assertSame(OperationOutcome::Failed, $result->outcome());
+        self::assertSame(OperationOutcome::Failed, $result->outcome()->outcome());
     }
 
     public function test_it_closes_both_connections_on_unbind(): void
@@ -121,7 +121,7 @@ public function test_it_does_not_forward_an_abandon(): void
             new AbandonRequest(2),
         ));
 
-        self::assertSame(OperationOutcome::Succeeded, $result->outcome());
+        self::assertSame(OperationOutcome::Succeeded, $result->outcome()->outcome());
     }
 
     private function contextFor(
diff --git a/tests/unit/Server/Proxy/ProxyRequestPipelineTest.php b/tests/unit/Server/Proxy/ProxyRequestPipelineTest.php
index fecbd0e7..4cd33274 100644
--- a/tests/unit/Server/Proxy/ProxyRequestPipelineTest.php
+++ b/tests/unit/Server/Proxy/ProxyRequestPipelineTest.php
@@ -59,7 +59,7 @@ public function test_it_forwards_everything_else(): void
         $this->forwarder
             ->expects(self::once())
             ->method('handle')
-            ->willReturn(OperationOutcomeResult::succeeded());
+            ->willReturn(ResponseStream::none(OperationOutcomeResult::succeeded()));
         $this->startTlsHandler
             ->expects(self::never())
             ->method('handleRequest');
diff --git a/tests/unit/Sync/Provider/SyncPersistStreamerTest.php b/tests/unit/Sync/Provider/SyncPersistStreamerTest.php
index 26b908dd..1e9e78b5 100644
--- a/tests/unit/Sync/Provider/SyncPersistStreamerTest.php
+++ b/tests/unit/Sync/Provider/SyncPersistStreamerTest.php
@@ -28,6 +28,10 @@
 use FreeDSx\Ldap\Protocol\Authorization\AuthzId;
 use FreeDSx\Ldap\Protocol\LdapMessageRequest;
 use FreeDSx\Ldap\Protocol\LdapMessageResponse;
+use FreeDSx\Ldap\Protocol\Queue\Response\Cancellation;
+use FreeDSx\Ldap\Protocol\Queue\Response\QueueWriterConfig;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream;
+use FreeDSx\Ldap\Protocol\Queue\Response\ResponseWriter;
 use FreeDSx\Ldap\Protocol\Queue\ServerQueue;
 use FreeDSx\Ldap\Search\Filters;
 use FreeDSx\Ldap\Server\AccessControl\AccessControlInterface;
@@ -39,6 +43,7 @@
 use FreeDSx\Ldap\Server\Backend\Storage\Journal\InMemoryChangeJournal;
 use FreeDSx\Ldap\Server\Backend\Storage\Journal\Read\ChangeStream;
 use FreeDSx\Ldap\Server\Backend\Storage\Journal\ReplicaId;
+use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult;
 use FreeDSx\Ldap\Server\Token\TokenInterface;
 use FreeDSx\Ldap\Sync\Provider\SyncPersistStreamer;
 use FreeDSx\Ldap\Sync\Provider\SyncResultProjector;
@@ -89,9 +94,13 @@ protected function setUp(): void
 
         $this->queue = $this->createMock(ServerQueue::class);
         $this->queue
-            ->method('sendMessage')
-            ->willReturnCallback(function (LdapMessageResponse ...$responses): ServerQueue {
+            ->method('sendMessages')
+            ->willReturnCallback(function (iterable $responses): ServerQueue {
                 foreach ($responses as $response) {
+                    if (!$response instanceof LdapMessageResponse) {
+                        continue;
+                    }
+
                     $this->sent[] = $response;
                 }
 
@@ -116,7 +125,6 @@ protected function setUp(): void
             ->willReturn(true);
 
         $this->subject = new SyncPersistStreamer(
-            queue: $this->queue,
             backend: $this->backend,
             projector: new SyncResultProjector(
                 accessControl: $accessControl,
@@ -132,7 +140,7 @@ public function test_a_cancel_closes_the_stream_with_a_canceled_result_and_ackno
     {
         $this->cancelSignals = [new LdapMessageRequest(9, new CancelRequest(5))];
 
-        $this->persist();
+        $this->drive($this->subject);
 
         $done = $this->firstSent(fn(LdapMessageResponse $m): bool => $m->getResponse() instanceof SearchResultDone);
         self::assertNotNull($done);
@@ -156,7 +164,7 @@ public function test_an_abandon_closes_the_stream_without_any_response(): void
     {
         $this->cancelSignals = [new LdapMessageRequest(9, new AbandonRequest(5))];
 
-        $this->persist();
+        $this->drive($this->subject);
 
         self::assertNull(
             $this->firstSent(fn(LdapMessageResponse $m): bool => $m->getResponse() instanceof SearchResultDone),
@@ -175,7 +183,7 @@ public function test_a_change_that_lands_during_the_loop_is_pushed_as_an_add():
             $this->cancelSignals[] = new LdapMessageRequest(9, new CancelRequest(5));
         };
 
-        $this->persist();
+        $this->drive($this->subject);
 
         $entry = $this->firstSent(fn(LdapMessageResponse $m): bool => $m->getResponse() instanceof SearchResultEntry);
         self::assertNotNull($entry);
@@ -207,7 +215,6 @@ public function test_a_trim_past_the_consumer_position_ends_with_refresh_require
             ->willReturn(new ReplicaId(self::ORIGIN));
 
         $subject = new SyncPersistStreamer(
-            queue: $this->queue,
             backend: $this->backend,
             projector: new SyncResultProjector(
                 accessControl: $this->createMock(AccessControlInterface::class),
@@ -218,13 +225,7 @@ public function test_a_trim_past_the_consumer_position_ends_with_refresh_require
             pollInterval: 0.0,
         );
 
-        $subject->persist(
-            0,
-            $this->request(),
-            $this->token,
-            5,
-            new Dn('dc=example,dc=com'),
-        );
+        $this->drive($subject);
 
         $done = $this->firstSent(fn(LdapMessageResponse $m): bool => $m->getResponse() instanceof SearchResultDone);
         self::assertNotNull($done);
@@ -236,14 +237,29 @@ public function test_a_trim_past_the_consumer_position_ends_with_refresh_require
         );
     }
 
-    private function persist(): void
+    /**
+     * Drives the persist generator through the writer, which polls the queue and feeds the cancellation token.
+     */
+    private function drive(SyncPersistStreamer $subject): void
     {
-        $this->subject->persist(
-            0,
-            $this->request(),
-            $this->token,
+        $cancellation = new Cancellation();
+        $stream = ResponseStream::streaming(
+            $subject->stream(
+                0,
+                $this->request(),
+                $this->token,
+                5,
+                new Dn('dc=example,dc=com'),
+                $cancellation,
+            ),
+            fn(): OperationOutcomeResult => OperationOutcomeResult::succeeded(),
+            new QueueWriterConfig(flushPerMessage: true, signalInterval: 1),
+            $cancellation,
+        );
+
+        (new ResponseWriter($this->queue))->write(
+            $stream,
             5,
-            new Dn('dc=example,dc=com'),
         );
     }