anchor.diff:        throw StateError('Cannot add new channels after shutdown() was called');
       }
   
  -    final comm = DriftCommunication(channel)..setRequestHandler(_handleRequest);
  +    final comm = DriftCommunication(channel);
  +    comm.setRequestHandler((request) => _handleRequest(comm, request));
  +
       _activeChannels.add(comm);
       comm.closed.then((_) => _activeChannels.remove(comm));
     }
  
positive.code:         connected.request(payload);
        }
      } else if (payload is RunTransactionAction) {
        return _transactionControl(payload.control, payload.executorId);
      } else if (payload is RequestCancellation) {
        _cancellableOperations[payload.originalRequestId]?.cancel();
        return null;
      }
    }
  
    Future<bool> _handleEnsureOpen(EnsureOpen open) async {
      _dbUser.schemaVersion = open.schemaVersion;
      final executor = await _loadExecutor(open.executorId);
  
      return await executor.ensureOpen(_dbUser);
    }
  
    Future<dynamic> _runQuery(StatementMethod method, String sql,
  
negative.code: 
  class _ServerDbUser implements QueryExecutorUser {
    final ServerImplementation _server;
    final DriftCommunication connection;
    @override
    final int schemaVersion;
  
    _ServerDbUser(this._server, this.connection, this.schemaVersion);
  
    @override
    Future<void> beforeOpen(
        QueryExecutor executor, OpeningDetails details) async {
      final id = _server._putExecutor(executor, beforeCurrent: true);
      try {
        await connection.request(RunBeforeOpen(details, id));
      } finally {
        _server._releaseExecutor(id);
      }
  